Pyspark: запись фрейма данных в таблицу Cassandra не работает - PullRequest
Новая
       53

Pyspark: запись фрейма данных в таблицу Cassandra не работает

0 голосов
/ 26 сентября 2019

Когда я пытался прочитать данные из таблицы Cassandra через pyspark работает нормально.Но когда я пытался записать Dataframe в таблицу Cassandra, он выдает java.lang.NoClassDefFoundError с теми же соединительными пакетами Spark-Cassandra.

Детали версий:

Cassandra: 

Connected to Test Cluster at 127.0.0.1:9042.
[cqlsh 5.0.1 | Cassandra 3.0.18 | CQL spec 3.4.0 | Native protocol v4]
Use HELP for help.

Spark: 

Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.4.3
      /_/

Using Python version 2.7.5 (default, Jun 11 2019 14:33:56)

Spark - Соединитель Cassandra: 

bin/pyspark --packages datastax:spark-cassandra-connector:2.4.0-s_2.11

Код: 

>>> from pyspark import SparkContext, SparkConf
>>> from pyspark.sql import SQLContext, SparkSession
>>> from pyspark.sql.types import *
>>> import os
>>> spark = SparkSession.builder \
...   .appName('SparkCassandraApp') \
...   .config('spark.cassandra.connection.host', '127.0.0.1') \
...   .config('spark.cassandra.connection.port', '9042') \
...   .config('spark.cassandra.output.consistency.level','ONE') \
...   .master('local[2]') \
...   .getOrCreate()
>>> df = spark.read.format("org.apache.spark.sql.cassandra").options(table="emp",keyspace="tutorialspoint").load()
>>> df.show()
+------+---------+--------+----------+-------+
|emp_id| emp_city|emp_name| emp_phone|emp_sal|
+------+---------+--------+----------+-------+
|     2|Hyderabad|   robin|9848022339|  40000|
|     1|Hyderabad|     ram|9848022338|  50000|
|     3|  Chennai|  rahman|9848022330|  45000|
+------+---------+--------+----------+-------+

В том же терминале при попытке записи в таблицу Cassandra 

>>> df.write\
... .format("org.apache.spark.sql.cassandra")\
... .mode('append')\
... .options(table="emp", keyspace="tutorialspoint")\
... .save()

19/09/26 15:34:15 ERROR Executor: Exception in task 6.0 in stage 3.0 (TID 25)
java.lang.NoClassDefFoundError: com/twitter/jsr166e/LongAdder
        at org.apache.spark.metrics.OutputMetricsUpdater$TaskMetricsSupport$class.$init$(OutputMetricsUpdater.scala:107)
        at org.apache.spark.metrics.OutputMetricsUpdater$TaskMetricsUpdater.<init>(OutputMetricsUpdater.scala:153)
        at org.apache.spark.metrics.OutputMetricsUpdater$.apply(OutputMetricsUpdater.scala:75)
        at com.datastax.spark.connector.writer.TableWriter.writeInternal(TableWriter.scala:209)
        at com.datastax.spark.connector.writer.TableWriter.insert(TableWriter.scala:197)
        at com.datastax.spark.connector.writer.TableWriter.write(TableWriter.scala:183)
        at com.datastax.spark.connector.RDDFunctions$$anonfun$saveToCassandra$1.apply(RDDFunctions.scala:36)
        at com.datastax.spark.connector.RDDFunctions$$anonfun$saveToCassandra$1.apply(RDDFunctions.scala:36)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
        at org.apache.spark.scheduler.Task.run(Task.scala:121)
        at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
19/09/26 15:34:15 ERROR Executor: Exception in task 7.0 in stage 3.0 (TID 26)
java.lang.NoClassDefFoundError: com/twitter/jsr166e/LongAdder
        at org.apache.spark.metrics.OutputMetricsUpdater$TaskMetricsSupport$class.$init$(OutputMetricsUpdater.scala:107)
        at org.apache.spark.metrics.OutputMetricsUpdater$TaskMetricsUpdater.<init>(OutputMetricsUpdater.scala:153)
        at org.apache.spark.metrics.OutputMetricsUpdater$.apply(OutputMetricsUpdater.scala:75)
        at com.datastax.spark.connector.writer.TableWriter.writeInternal(TableWriter.scala:209)
        at com.datastax.spark.connector.writer.TableWriter.insert(TableWriter.scala:197)
        at com.datastax.spark.connector.writer.TableWriter.write(TableWriter.scala:183)
        at com.datastax.spark.connector.RDDFunctions$$anonfun$saveToCassandra$1.apply(RDDFunctions.scala:36)
        at com.datastax.spark.connector.RDDFunctions$$anonfun$saveToCassandra$1.apply(RDDFunctions.scala:36)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
        at org.apache.spark.scheduler.Task.run(Task.scala:121)
        at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

1 Ответ

1 голос
/ 27 сентября 2019

Вам нужен другой cassandra-connector.Соединитель Datastax будет работать для Scala / Java, но вам нужно что-то для аналога Python.

argumentsenot / pyspark-cassandra - это порт Python соединителя Datastax Spark.

Пример: 

import pyspark_cassandra

conf = SparkConf() \
    .setAppName("PySpark Cassandra Test") \
    .setMaster("spark://spark-master:7077") \
    .set("spark.cassandra.connection.host", "cas-1")

sc = CassandraSparkContext(conf=conf)

rdd = sc.parallelize([{
    "key": k,
    "stamp": datetime.now(),
    "val": random() * 10,
    "tags": ["a", "b", "c"],
    "options": {
        "foo": "bar",
        "baz": "qux",
    }
} for k in ["x", "y", "z"]])

rdd.saveToCassandra(
    "keyspace",
    "table",
    ttl=timedelta(hours=1),
)
...