Spark UDF per cercare le chiavi utilizzando il connettore Cassandra
Sto provando a utilizzare cassandra come archivio di ricerca di valori chiave in alcuni dei nostri lavori scintilla.
Utilizziamo principalmente i dataframe e ci siamo allontanati dalle API RDD.
Invece di unirmi ai tavoli, caricarli in spark o
spingere il join a cassandra e prendere misure per evitare
scansioni di tavoli di grandi dimensioni, ho pensato di poter semplicemente scrivere un UDF Spark che si connette a cassandra e cerca una chiaveVoglio inoltre convertire la riga del risultato in un oggetto di classe case e restituire l'oggetto.
Ho ottenuto alcune di queste informazioni sulla base delle risposte a questa domanda di seguito. withSessionDo riutilizza una sessione a livello di JVM sottostante che è disponibile su ogni nodo Utilizzo corretto di Spark Cassandra Connector
val connector = CassandraConnector(sparkConf) // I Know this is serializable.
def lookupKey(connector: CassandraConnector, keyspace: String, table: String): UserDefineFunction = udf((key: String) => {
connector.withSessionDo(session => {
val stmt = session.prepare(s"SELECT * FROM $keyspace.$table WHERE key = ?")
val result = session.execute( stmt.bind(key) )
MyCaseClass(
fieldl1 = result.getString(0),
fieldl2 = result.getInt(1)
...
)
}
})
La sessione non è serializzabile, quindi non possiamo crearne una al di fuori di udf e passarla in modo da poter utilizzare il gestore di mappatura per convertire le righe in istanze della classe case. Un approccio alternativo utilizzando Mapping Manager,
def lookupKeyAlt(connector: CassandraConnector, keyspace: String, table: String): UserDefineFunction = udf((key: String) => {
connector.withSessionDo(session => {
val manager = new MappingManager(session) // session isn't serializable, so creating one outside and passing to udf is not an option if wf we were willing to do the session management.
val mapperClass = manager.mapper(classOf[MyCaseClass], keyspace)
mapperClass.get(key)
}
})
Sono nuovo di cassandra quindi per favore abbi pazienza con me su alcune domande.
- Ci sono dei trucchi in questi approcci di cui non sono a conoscenza?
- Nel secondo approccio, capisco che stiamo creando un nuovo MappingManager (sessione) con ogni chiamata dell'UDF. Questo utilizzerà ancora la sessione di livello jvm e aprirà altre sessioni? È anche giusto istanziare MappingManager con ogni chiamata? La sessione non è serializzabile, quindi non posso crearla all'esterno e passarla all'UDF.
- Quali sono gli altri modi per convertire una riga di risultato in un oggetto di una classe di casi?
- Ci sono alternative migliori per fare questo tipo di ricerca?
Risposte
Stai cercando di emulare ciò che Spark Cassandra Connector (SCC) sta facendo sotto il cofano, ma la tua implementazione sarà molto più lenta di quella di SCC perché stai utilizzando l'API sincrona e ottieni tutti i dati uno dopo l'altro, mentre SCC utilizza l'API asincrona ed estrarre i dati per più righe in parallelo.
Il modo migliore per ottenere ciò che desideri è utilizzare l'unione ottimizzata per Cassandra (spesso chiamata "unione diretta"). Questo tipo di join è sempre stato disponibile per l'API RDD, ma per molto tempo è stato disponibile per l'API Dataframe solo nella versione commerciale del connettore. Ma a partire da SCC 2.5.0 ( rilasciato a maggio 2020 ), questa funzionalità è disponibile anche in versione open source, quindi puoi usarla invece di crearne l'emulazione. L'unione diretta viene eseguita solo quando si abilitano estensioni speciali di Catalyst , passando il filespark.sql.extensions=com.datastax.spark.connector.CassandraSparkExtensions
durante la configurazione di SparkSession (ad esempio tramite la riga di comando). Successivamente, puoi eseguire l'unione con la tabella Cassandra tramite chiave primaria completa o parziale e SCC convertirà automaticamente l'unione in singole richieste a Cassandra che vengono eseguite in modo molto efficace. Puoi verificare che ciò avvenga eseguendo explain
sul dataframe unito, quindi dovresti vedere qualcosa del genere (cerca la stringa Cassandra Direct Join ):
scala> joined.explain
== Physical Plan ==
Cassandra Direct Join [pk = id#30, c1 = cc1#32] test.jtest1 - Reading (pk, c1, c2, v) Pushed {}
+- *(1) Project [cast(id#28L as int) AS id#30, cast(id#28L as int) AS cc1#32]
+- *(1) Range (1, 5, step=1, splits=8)
Di recente ho scritto un lungo post sul blog che spiega come eseguire join efficaci con i dati in Cassandra utilizzando entrambe le API Dataframe e RDD: non voglio ripeterlo qui :-)