火花的作业（斯卡拉）写类型日期卡桑德拉

我使用DSE 5.1（火花2.0.2.6和卡桑德拉3.10.0.1652）火花的作业（斯卡拉）写类型日期卡桑德拉

我卡桑德拉表：

CREATE TABLE ks.tbl (
    dk int, 
    date date, 
    ck int, 
    val int, 
PRIMARY KEY (dk, date, ck) 
) WITH CLUSTERING ORDER BY (date DESC, ck ASC);

数据如下：

dk | date  | ck | val 
----+------------+----+----- 
    1 | 2017-01-01 | 1 | 100 
    1 | 2017-01-01 | 2 | 200

我的代码必须读取这些数据，并写同样的事情，但与昨天的日期（它编译成功）：

package com.datastax.spark.example 

import com.datastax.spark.connector._ 
import com.datastax.spark.connector.cql.CassandraConnector 
import org.apache.spark.sql.hive.HiveContext 
import org.apache.spark.{SparkConf, SparkContext} 
import com.github.nscala_time.time._ 
import com.github.nscala_time.time.Imports._ 

object test extends App { 

    val conf = new SparkConf().setAppName("DSE calculus app TEST") 
    val sc = new SparkContext(conf) 

    val yesterday= (DateTime.now - 1.days).toString(StaticDateTimeFormat.forPattern("yyyy-MM-dd")) 

    val tbl = sc.cassandraTable("ks","tbl").select("dk","date","ck","val").where("dk=1") 

    tbl.map(row => (row.getInt("dk"),yesterday,row.getInt("ck"),row.getInt("val"))).saveToCassandra("ks","tbl") 

    sc.stop() 
    sys.exit(0) 
}

当我运行这个程序：

dse spark-submit --class com.datastax.spark.example.test test-assembly-0.1.jar

它不能正确写入卡桑德拉。看来日期变量没有正确插入地图中。我得到的错误是：

Error: 
WARN 2017-05-08 22:23:16,472 org.apache.spark.scheduler.TaskSetManager: Lost task 0.0 in stage 0.0 (TID 0, <IP of one of my nodes>): java.io.IOException: Failed to write statements to ks.tbl. 
     at com.datastax.spark.connector.writer.TableWriter$$anonfun$writeInternal$1.apply(TableWriter.scala:207) 
     at com.datastax.spark.connector.writer.TableWriter$$anonfun$writeInternal$1.apply(TableWriter.scala:175) 
     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:175) 
     at com.datastax.spark.connector.writer.TableWriter.insert(TableWriter.scala:162) 
     at com.datastax.spark.connector.writer.TableWriter.write(TableWriter.scala:149) 
     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:70) 
     at org.apache.spark.scheduler.Task.run(Task.scala:86) 
     at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:274) 
     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
     at java.lang.Thread.run(Thread.java:748)

然而，当我直接在地图语句插入日期（字符串）如下代码没有正确地插入数据：

tbl.map(row => (row.getInt("dk"),"2017-02-02",row.getInt("ck"),row.getInt("val"))).saveToCassandra("ks","tbl")

还插入数据正确，如果我昨天设置为一个整数（自纪元以来的天数）。这将是最佳的，但不能得到'昨天'表现这种方式

编辑：这不会正确插入数据，实际上。无论将“昨天”设置为1还是100,000,000，它始终插入纪元（'1970-01-01）

失败的代码表现正确，并且正如我在DSE Spark控制台中所期望的那样。

我只是无法弄清楚我做错了什么。欢迎任何帮助。

编辑2：excecutor 0 stderr log确实表明它试图在列日期插入一个空值，这显然不可能，因为它是一个聚类列。

来源

2017-05-08 Mematematica

您需要发布执行日志。由于驱动程序只能看到“写入语句失败”，因此它不会导致造成它的个别尝试和失败。 – RussS

你的意思是来自Spark主UI中应用程序的所有执行者的标准输出stderr？ – Mematematica

至少你得到了例外:) :) – RussS

在为Spark作业编写代码时，了解何时设置了特定变量以及何时序列化是很重要的。让我们来看看一个音符从App特质文档

注意事项

应当指出的是，这种特性是使用 DelayedInit功能，这意味着该对象的字段将不会一直执行在主方法执行之前初始化。

这意味着在实际运行代码时，可能不会在Executors上初始化对App正文中使用的变量的引用。

我的猜测是，你写的lambda包含一个val的引用，它是在App类的Delayed init部分初始化的。这意味着执行程序上未运行Main方法的代码的序列化版本会获取该值的未初始化版本（空值）。

将常量切换到lazy val（或将其移动到单独的对象或类中）可以通过确保该值远程初始化（lazy val）或简单地序列化初始化（单独的类/对象）来解决此问题。

来源

2017-05-09 16:01:53 RussS

我想我知道你的问题是什么。
您可能会看到完整的日志文件。您只需附加其中的一部分...
今天有类似的错误，当创建keyspace与replication_factor：3时，我只有一个cassandra实例。

所以我改变它，问题不复存在。

ALTER KEYSPACE "some_keyspace_name" WITH REPLICATION = 
    { 'class' : 'SimpleStrategy', 'replication_factor' : 1 };

Here is my error.log file

和日志的重要组成部分：

Logging.scala[logError]:72) - Failed to execute: [email protected] 
com.datastax.driver.core.exceptions.UnavailableException: Not enough replicas available for query at consistency LOCAL_QUORUM (2 required but only 1 alive)

来源

2017-08-03 23:34:58

火花的作业（斯卡拉）写类型日期卡桑德拉

回答

相关问题