0

我有一个带有 spark-cassandra-connector 1.6.2 的数据框设置。我尝试用 cassandra 进行一些转换。Datastax 企业版是 5.0.5。

DataFrame df1 =  sparkContext
            .read().format("org.apache.spark.sql.cassandra")
            .options(readOptions).load()
            .where("field2 ='XX'")
            .limit(limitVal)
            .repartition(partitions);

List<String> distinctKeys = df1.getColumn("field3").collect();  

values = some transformations to get IN query values;

String cassandraQuery = String.format("SELECT * FROM "
            + "table2 "
            + "WHERE field2 = 'XX' "
            + "AND field3 IN (%s)", values);
DataFrame df2 = sparkContext.cassandraSql(cassandraQuery);

String column1 = "field3";
String column2 = "field4";
List<String> columns = new ArrayList<>();
        columns.add(column1);
        columns.add(column2);
scala.collection.Seq<String> usingColumns = 
scala.collection.JavaConverters.
collectionAsScalaIterableConverter(columns).asScala().toSeq();
DataFrame joined = df1.join(df2, usingColumns, "left_outer");

List<Row> collected = joined.collectAsList(); // doestn't work
Long count = joined.count(); // works

这是异常日志，看起来spark正在创建cassandra源代码，并且无法序列化。

java.io.NotSerializableException: java.util.ArrayList$Itr
Serialization stack:
- object not serializable (class: 
org.apache.spark.sql.cassandra.CassandraSourceRelation, value:  
org.apache.spark.sql.cassandra.CassandraSourceRelation@1c11a496)
- field (class: org.apache.spark.sql.execution.datasources.LogicalRelation, 
name: relation, type: class org.apache.spark.sql.sources.BaseRelation)
- object (class org.apache.spark.sql.execution.datasources.LogicalRelation, 
Relation[fields] 
org.apache.spark.sql.cassandra.CassandraSourceRelation@1c11a496 
)
- field (class: org.apache.spark.sql.catalyst.plans.logical.Filter, name: 
child, type: class org.apache.spark.sql.catalyst.plans.logical.LogicalPlan)
- object (class org.apache.spark.sql.catalyst.plans.logical.Filter, Filter 
(field2#0 = XX)
+- Relation[fields] 
org.apache.spark.sql.cassandra.CassandraSourceRelation@1c11a496
)
- field (class: org.apache.spark.sql.catalyst.plans.logical.Repartition, name: 
child, type: class org.apache.spark.sql.catalyst.plans.logical.LogicalPlan)
- object (class org.apache.spark.sql.catalyst.plans.logical.Repartition, 
Repartition 4, true
+- Filter (field2#0 = XX)
+- Relation[fields] 
org.apache.spark.sql.cassandra.CassandraSourceRelation@1c11a496
)
- field (class: org.apache.spark.sql.catalyst.plans.logical.Join, name: left, 
type: class org.apache.spark.sql.catalyst.plans.logical.LogicalPlan)
- object (class org.apache.spark.sql.catalyst.plans.logical.Join, Join 
LeftOuter, Some(((field3#2 = field3#18) && (field4#3 = field4#20)))
:- Repartition 4, true
:  +- Filter (field2#0 = XX)
:     +- Relation[fields] 
org.apache.spark.sql.cassandra.CassandraSourceRelation@1c11a496
+- Project [fields]
+- Filter ((field2#17 = YY) && field3#18 IN (IN array))
  +- Relation[fields] 
org.apache.spark.sql.cassandra.CassandraSourceRelation@7172525e
)
- field (class: org.apache.spark.sql.catalyst.plans.logical.Project, name: 
child, type: class org.apache.spark.sql.catalyst.plans.logical.LogicalPlan)
- object (class org.apache.spark.sql.catalyst.plans.logical.Project, Project 
[fields]
+- Join LeftOuter, Some(((field3#2 = field3#18) && (field4#3 = field4#20)))
:- Repartition 4, true
:  +- Filter (field2#0 = XX)
:     +- Relation[fields] 
org.apache.spark.sql.cassandra.CassandraSourceRelation@1c11a496
+- Project [fields]
  +- Filter ((field2#17 = XX) && field3#18 IN (IN array))
     +- Relation[fields] 
org.apache.spark.sql.cassandra.CassandraSourceRelation@7172525e
)
- field (class: org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4, name: 
$outer, type: class org.apache.spark.sql.catalyst.trees.TreeNode)
- object (class org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4, 
<function1>)
- field (class: 
org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4$$anonfun$apply$9, 
name: $outer, type: class 
org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4)
- object (class 
org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4$$anonfun$apply$9, 
<function1>)
- field (class: scala.collection.immutable.Stream$$anonfun$map$1, name: f$1, 
type: interface scala.Function1)
- object (class scala.collection.immutable.Stream$$anonfun$map$1, <function0>)
- writeObject data (class: scala.collection.immutable.$colon$colon)
- object (class scala.collection.immutable.$colon$colon, 
List(org.apache.spark.OneToOneDependency@17f43f4a))
- field (class: org.apache.spark.rdd.RDD, name: 
org$apache$spark$rdd$RDD$$dependencies_, type: interface scala.collection.Seq)
- object (class org.apache.spark.rdd.MapPartitionsRDD, MapPartitionsRDD[32] at 
collectAsList at RevisionPushJob.java:308)
- field (class: org.apache.spark.rdd.RDD$$anonfun$collect$1, name: $outer, 
type: class org.apache.spark.rdd.RDD)
- object (class org.apache.spark.rdd.RDD$$anonfun$collect$1, <function0>)
- field (class: org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12, name: 
$outer, type: class org.apache.spark.rdd.RDD$$anonfun$collect$1)
- object (class org.apache.spark.rdd.RDD$$anonfun$collect$1$$anonfun$12, 
<function1>)

是否可以使其序列化？为什么计数操作有效但收集操作无效？

更新：

回到它之后，事实证明，在 Java 中，您首先将 Java Iterable 转换为 scala 缓冲区并从中创建一个 scala Iterable -> Seq。否则它不起作用。感谢 Russel 让我注意到问题的原因。

String attrColumn1 = "column1";
            String attrColumn2 = "column2";
            String attrColumn3 = "column3";
            String attrColumn4 = "column4";
            List<String> attrColumns = new ArrayList<>();
            attrColumns.add(attrColumn1);
            attrColumns.add(attrColumn2);
            attrColumns.add(attrColumn3);
            attrColumns.add(attrColumn4);
            Seq<String> usingAttrColumns = 
JavaConverters.asScalaBufferConverter(attrColumns).asScala().toList();
4

1 回答 1

0

请参阅指向java.util.ArrayList$Itr您的不可序列化位的错误消息，我认为这可能是对

 List<String> columns = new ArrayList<>();
    columns.add(column1);
    columns.add(column2);

哪个隐式转换可能需要对数组列表迭代器进行序列化？这是我看到的唯一 ArrayList 所以它可能是罪魁祸首。它也可能在您为“值”删除的代码中。

当你这样做时，Count它可以丢弃列信息，这样可能会节省你，但我不能确定。

所以 TLDR 我的建议是尝试从代码中删除一些东西，然后重新替换和构建你的代码以找到不可序列化的位。

于 2017-09-07T04:27:01.353 回答