java.util.NoSuchElementException：在表 abc.company_vals 中找不到列：companyId，companyName

Question

使用库、spark-cassandra-connector_2-11.jar 和 spark-sql-2.4.1.jar

有如下Cassandra表

CREATE TABLE abc.company_vals(
    companyId int,
    companyName text,
    year int,
    quarter text,
    revenue int,
    PRIMARY KEY (companyId, year)
) WITH CLUSTERING ORDER BY ( year DESC );

尝试使用以下火花结构化流将数据插入到上面

List<Row> data  = Arrays.asList(
                    RowFactory.create(10002 , "TCS",2004,"Q4",7800),
                    RowFactory.create(10003, "GE",2004,"Q4",7800),
                    RowFactory.create(10004, "Oracle",2004,"Q4",7800),
                    RowFactory.create(10005, "epam",2004,"Q4",7800),
                    RowFactory.create(10006, "Dhfl",2004,"Q4",7800),
                    RowFactory.create(10007, "Infosys",2004,"Q4",7800)
               )

StructType schema = new StructType()
                      .add("companyId", DataTypes.IntegerType)
                      .add("companyName", DataTypes.StringType)
                      .add("year", DataTypes.IntegerType)
                      .add("quarter", DataTypes.StringType)
                      .add("revenue", DataTypes.IntegerType);

             Dataset<Row> companyDf = sparkSession.createDataFrame(data, schema).toDF();


             companyDf
             .write()
             .format("org.apache.spark.sql.cassandra")
                .option("table","company_vals")
                .option("keyspace",  "abc")
                .mode(SaveMode.Append)
                .save();

我更改了表的顺序，例如 pk 、簇键和其余列，相应地更改了 StructType 和输入...但仍然是同样的错误。

出现错误：

java.util.NoSuchElementException: Columns not found in table abc.company_vals: companyId, companyName
at com.datastax.spark.connector.SomeColumns.selectFrom(ColumnSelector.scala:44)
at com.datastax.spark.connector.writer.TableWriter$.apply(TableWriter.scala:385)
at com.datastax.spark.connector.RDDFunctions.saveToCassandra(RDDFunctions.scala:35)
at org.apache.spark.sql.cassandra.CassandraSourceRelation.insert(CassandraSourceRelation.scala:76)
at org.apache.spark.sql.cassandra.DefaultSource.createRelation(DefaultSource.scala:86)
at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:45)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:70)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:68)
at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:86)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:80)
at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:80)
at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:668)
at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:668)
at org.apache.spark.sql.execution.SQLExecution$$anonfun$withNewExecutionId$1.apply(SQLExecution.scala:78)
at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:125)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:73)
at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:668)
at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:276)
at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:270)

我在这里做错了什么？如何解决这个问题？

Answer 1

问题在于 Spark 连接器使用区分大小写的名称，而在 CQL 中它们不区分大小写，直到将列名称放入双引号中。因此，您需要将表中的字段分别声明为区分大小写的 "companyId" 和 "companyName"，或者在 Spark 应用中使用小写名称。