java.util.NoSuchElementException: столбцы не найдены в таблице abc.company_vals: companyId, companyName - PullRequest
Новая
       32

java.util.NoSuchElementException: столбцы не найдены в таблице abc.company_vals: companyId, companyName

1 голос
/ 31 мая 2019

Использование libs, 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 & input соответственно ... но все еще та же ошибка.

Получение ошибки: 

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)

Что я здесь не так делаю?как это исправить?

1 Ответ

3 голосов
/ 31 мая 2019

Проблема в том, что Spark Connector использует имена, чувствительные к регистру, тогда как в CQL они нечувствительны к регистру, пока имена столбцов не будут заключены в двойные кавычки. Поэтому вам нужно либо объявить поля в таблице с учетом регистра как "companyId" и "companyName" соответственно, либо использовать имена в нижнем регистре в приложении Spark.

...