Come eseguire l'inizializzazione nella scintilla?

Question

Voglio eseguire ricerche geoip dei miei dati in spark. Per farlo uso il database geoIP di MaxMind.

Quello che voglio fare è inizializzare un oggetto di database geoip una volta su ogni partizione, e in seguito usarlo per cercare la città relativa ad un indirizzo IP.

La scintilla ha una fase di inizializzazione per ogni nodo, o dovrei invece verificare se una variabile di istanza non è definita, e in caso affermativo, inizializzarla prima di continuare? Per esempio. qualcosa di simile (questo è python ma voglio una soluzione Scala):

class IPLookup(object): 
    database = None 

    def getCity(self, ip): 
     if not database: 
     self.database = self.initialise(geoipPath) 
    ... 

Naturalmente, facendo questo richiede scintilla puntate tutto l'oggetto, cosa che la documentazione in guardia contro.

Answer 1

Questo sembra un buon utilizzo di una variabile di trasmissione. Hai esaminato la documentazione per quella funzionalità e se lo hai non riesce a soddisfare le tue esigenze in qualche modo?

Answer 2

In Spark, per partizione operazioni possono essere do utilizzando:

def mapPartitions[U](f: (Iterator[T]) ⇒ Iterator[U], preservesPartitioning: Boolean = false) 

Questa mappatore eseguirà la funzione f volta per partizione su un iteratore di elementi. L'idea è che il costo di impostazione delle risorse (come le connessioni DB) sarà compensato con l'utilizzo di tali risorse su un numero di elementi nell'iteratore.

Esempio:

val logsRDD = ??? 
logsRDD.mapPartitions{iter => 
    val geoIp = new GeoIPLookupDB(...) 
    // this is local map over the iterator - do not confuse with rdd.map 
    iter.map(elem => (geoIp.resolve(elem.ip),elem)) 
} 

Answer 3

Come @bearrito accennato - è possibile utilizzare caricare il GeoDB e poi trasmetterlo dal driver. Un'altra opzione da considerare è quella di fornire un servizio esterno che è possibile utilizzare per effettuare una ricerca. Potrebbe essere una cache in memoria come Redis/Memcached/Tacheyon o un datastore normale.