从Spark到HBase:org写作。阿帕奇。火花SparkException:任务不可序列化

Question

我在我大学的热图项目中，我们必须从txt文件（坐标、高度）中获取一些数据（212Go），然后将其放入HBase以在带有Express的Web客户端上检索它。

我练习使用144Mo文件，这是工作：

SparkConf conf = new SparkConf().setAppName("PLE");
JavaSparkContext context = new JavaSparkContext(conf);
JavaRDD<String> data = context.textFile(args[0]);
Connection co = ConnectionFactory.createConnection(getConf());
createTable(co);
Table table = co.getTable(TableName.valueOf(TABLE_NAME));
Put put = new Put(Bytes.toBytes("KEY"));

for (String s : data.collect()) {
    String[] tmp = s.split(",");
    put.addImmutable(FAMILY,
                    Bytes.toBytes(tmp[2]),
                    Bytes.toBytes(tmp[0]+","+tmp[1]));
}

table.put(put);

但是我现在使用212Go文件，我有一些内存错误，我猜收集方法会收集内存中的所有数据，所以212Go太多了。

所以现在我在尝试这个：

SparkConf conf = new SparkConf().setAppName("PLE");
JavaSparkContext context = new JavaSparkContext(conf);
JavaRDD<String> data = context.textFile(args[0]);
Connection co = ConnectionFactory.createConnection(getConf());
createTable(co);
Table table = co.getTable(TableName.valueOf(TABLE_NAME));
Put put = new Put(Bytes.toBytes("KEY"));

data.foreach(line ->{
    String[] tmp = line.split(",");
    put.addImmutable(FAMILY,
                    Bytes.toBytes(tmp[2]),
                    Bytes.toBytes(tmp[0]+","+tmp[1]));
});

table.put(put);

我得到了“org.apache.spark.SparkException:Task not serializable”，我搜索了它并尝试了一些修复，但没有成功，我在这里读到了：Task not serializable:java。伊奥。仅在类而非对象上调用闭包外函数时NotSerializableException

其实我并不完全理解这个话题，我只是一个学生，也许我的问题的答案是显而易见的，也许不是，无论如何，提前谢谢！

Answer 1

根据经验，序列化数据库连接（任何类型）都没有意义。无论是否有火花，都没有设计成序列化和反序列化。

为每个分区创建连接：

data.foreachPartition(partition -> {
  Connection co = ConnectionFactory.createConnection(getConf());
  ... // All required setup
  Table table = co.getTable(TableName.valueOf(TABLE_NAME));
  Put put = new Put(Bytes.toBytes("KEY"));
   while (partition.hasNext()) {
     String line = partition.next();
     String[] tmp = line.split(",");
     put.addImmutable(FAMILY,
                Bytes.toBytes(tmp[2]),
                Bytes.toBytes(tmp[0]+","+tmp[1]));
   }
   ... // Clean connections
});

我还建议阅读官方Spark Streaming编程指南中有关使用foreachRDD的设计模式。