我正在尝试使用 Spark 处理大型 cassandra 表(约 4.02 亿条目和 84 列),但得到的结果不一致。最初的要求是将一些列从该表复制到另一个表。复制数据后,我注意到新表中的一些条目丢失了。为了验证我是否对大型源表进行了计数,但每次我都得到不同的值。我在一个较小的表(约 700 万条记录)上尝试了查询,结果很好。
最初,我尝试使用 pyspark 进行计数。这是我的 pyspark 脚本:
spark = SparkSession.builder.appName("Datacopy App").getOrCreate()
df = spark.read.format("org.apache.spark.sql.cassandra").options(table=sourcetable, keyspace=sourcekeyspace).load().cache()
df.createOrReplaceTempView("data")
query = ("select count(1) from data " )
vgDF = spark.sql(query)
vgDF.show(10)
Spark提交命令如下:
~/spark-2.1.0-bin-hadoop2.7/bin/spark-submit --master spark://10.128.0.18:7077 --packages datastax:spark-cassandra-connector:2.0.1-s_2.11 --conf spark.cassandra.connection.host="10.128.1.1,10.128.1.2,10.128.1.3" --conf "spark.storage.memoryFraction=1" --conf spark.local.dir=/media/db/ --executor-memory 10G --num-executors=6 --executor-cores=2 --total-executor-cores 18 pyspark_script.py
上述 Spark 提交过程大约需要 90 分钟才能完成。我运行了三遍,这是我得到的计数:
- Spark迭代1:402273852
- Spark迭代2:402273884
- Spark迭代3:402274209
Spark在整个过程中没有显示任何错误或异常。我在 cqlsh 中运行相同的查询三次,并再次得到不同的结果:
- Cqlsh迭代1:402273598
- Cqlsh迭代2:402273499
- Cqlsh迭代3:402273515
我无法找出为什么我从同一查询中得到不同的结果。 Cassandra 系统日志 (/var/log/cassandra/system.log) 仅显示一次以下错误消息:
ERROR [SSTableBatchOpen:3] 2018-02-27 09:48:23,592 CassandraDaemon.java:226 - Exception in thread Thread[SSTableBatchOpen:3,5,main]
java.lang.AssertionError: Stats component is missing for sstable /media/db/datakeyspace/sensordata1-acfa7880acba11e782fd9bf3ae460699/mc-58617-big
at org.apache.cassandra.io.sstable.format.SSTableReader.open(SSTableReader.java:460) ~[apache-cassandra-3.9.jar:3.9]
at org.apache.cassandra.io.sstable.format.SSTableReader.open(SSTableReader.java:375) ~[apache-cassandra-3.9.jar:3.9]
at org.apache.cassandra.io.sstable.format.SSTableReader$4.run(SSTableReader.java:536) ~[apache-cassandra-3.9.jar:3.9]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[na:1.8.0_131]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[na:1.8.0_131]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) ~[na:1.8.0_131]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [na:1.8.0_131]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_131]
版本:
- 卡桑德拉 3.9。
- 火花2.1.0。
- Datastax 的 Spark-cassandra-connector 2.0.1
- 斯卡拉版本 2.11
Cluster:
- Spark 设置有 3 个工作节点和 1 个主节点。
- 3 个工作节点还安装了 cassandra 集群。
- 每个工作节点有 8 个 CPU 核心和 40 GB RAM。
任何帮助将不胜感激。