Spark SQL在Cassandra表之间传输数据 [英] spark sql to transfer data between Cassandra tables
问题描述
请在下面找到Cassandra表.
Please find the Cassandra table below.
我试图将数据从1个Cassandra表复制到另一个具有相同结构的Cassandra表中.
I am trying to copy data from 1 Cassandra table to another Cassandra table with same structure.
请帮助我.
CREATE TABLE data2 (
d_no text,
d_type text,
sn_perc int,
tse_dt timestamp,
f_lvl text,
ign_f boolean,
lk_loc text,
lk_ts timestamp,
mi_rem text,
nr_fst text,
perm_stat text,
rec_crt_dt timestamp,
sr_stat text,
sor_query text,
tp_dat text,
tp_ts timestamp,
tr_rem text,
tr_type text,
PRIMARY KEY (device_serial_no, device_type)
) WITH CLUSTERING ORDER BY (device_type ASC)
使用以下数据插入数据
Insert into data2(all column names) values('64FCFCFC','HUM',4,'1970-01-02 05:30:00’ ,’NA’,true,'NA','1970-01-02 05:40:00',’NA’,'NA','NA','1970-02-01 05:30:00','NA','NA','NA','1970-02-03 05:30:00','NA','NA');
注意: 当我尝试像'1970-01-02 05:30:00'这样插入的第4列时间戳，并且在dtaframe中也正确插入了时间戳，但是当从dataframe插入cassandra并使用select * from table时，我看到它是像1970-01-02 00:00:00.000000 + 0000这样插入
Note: The 4th column timestamp when i try to insert like this '1970-01-02 05:30:00’ ,and in dtaframe also timestamp inserted correctly ,but when insert from dataframe to cassandra and use select * from table, i see its being inserted like 1970-01-02 00:00:00.000000+0000
pom.xml
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.3.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-sql -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.3.1</version>
</dependency>
<!-- https://mvnrepository.com/artifact/com.datastax.spark/spark-cassandra-connector -->
<dependency>
<groupId>com.datastax.spark</groupId>
<artifactId>spark-cassandra-connector_2.11</artifactId>
<version>2.3.1</version>
</dependency>
我想读取这些值，并使用spark Scala将其写入另一个Cassandra表中.参见下面的代码:
I want to read these values and write it into another Cassandra table using spark Scala. See code below:
val df2 = spark.read
.format("org.apache.spark.sql.cassandra")
.option("spark.cassandra.connection.host","hostname")
.option("spark.cassandra.connection.port","9042")
.option( "spark.cassandra.auth.username","usr")
.option("spark.cassandra.auth.password","pas")
.option("keyspace","hr")
.option("table","data2")
.load()
Val df3 =doing some processing on df2.
df3.write
.format("org.apache.spark.sql.cassandra")
.mode("append")
.option("spark.cassandra.connection.host","hostname")
.option("spark.cassandra.connection.port","9042")
.option( "spark.cassandra.auth.username","usr")
.option("spark.cassandra.auth.password","pas")
.option("spark.cassandra.output.ignoreNulls","true")
.option("confirm.truncate","true")
.option("keyspace","hr")
.option("table","data3")
.save()
但是当我尝试使用上面的代码插入数据时，我遇到了错误
But i am getting below error, when i try to insert data using above code,
java.lang.IllegalArgumentException: requirement failed: Invalid row size: 18 instead of 17.
at scala.Predef$.require(Predef.scala:224)
at com.datastax.spark.connector.writer.SqlRowWriter.readColumnValues(SqlRowWriter.scala:23)
at com.datastax.spark.connector.writer.SqlRowWriter.readColumnValues(SqlRowWriter.scala:12)
at com.datastax.spark.connector.writer.BoundStatementBuilder.bind(BoundStatementBuilder.scala:99)
at com.datastax.spark.connector.writer.GroupingBatchBuilder.next(GroupingBatchBuilder.scala:106)
at com.datastax.spark.connector.writer.GroupingBatchBuilder.next(GroupingBatchBuilder.scala:31)
at scala.collection.Iterator$class.foreach(Iterator.scala:891)
at com.datastax.spark.connector.writer.GroupingBatchBuilder.foreach(GroupingBatchBuilder.scala:31)
at com.datastax.spark.connector.writer.TableWriter$$anonfun$writeInternal$1.apply(TableWriter.scala:233)
at com.datastax.spark.connector.writer.TableWriter$$anonfun$writeInternal$1.apply(TableWriter.scala:210)
at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$withSessionDo$1.apply(CassandraConnector.scala:112)
at com.datastax.spark.connector.cql.CassandraConnector$$anonfun$withSessionDo$1.apply(CassandraConnector.scala:111)
at com.datastax.spark.connector.cql.CassandraConnector.closeResourceAfterUse(CassandraConnector.scala:145)
at com.datastax.spark.connector.cql.CassandraConnector.withSessionDo(CassandraConnector.scala:111)
at com.datastax.spark.connector.writer.TableWriter.writeInternal(TableWriter.scala:210)
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:87)
at org.apache.spark.scheduler.Task.run(Task.scala:109)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
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)
推荐答案
That's a known problem (SPARKC-541) - you're copying the data from table that has DSE Search enabled to the table without it. You simply need to drop this column as part of your transformations:
val df3 = df2.drop("solr_query").... // your transformations
或者您可以简单地使用更新的驱动程序(如果使用的是OSS驱动程序，则为2.3.1)或包含此修复程序的相应DSE版本.
Or you can simply use the newer driver (2.3.1 if you're using OSS driver), or corresponding DSE release that contains this fix.
这篇关于Spark SQL在Cassandra表之间传输数据的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
