hadoop - 写elasticsearch时Spark不支持arraylist吗？

Question

我有以下结构:

mylist = [{"key1":"val1"}, {"key2":"val2"}]
myrdd = value_counts.map(lambda item: ('key', { 
    'field': somelist 
}))

我得到错误:15/02/10 15:54:08 INFO scheduler.TaskSetManager:在执行程序 ip-10-80-15-145.ec2.internal 上的 2.0 阶段 (TID 6) 丢失任务 1.0:org.apache.spark.SparkException(数据不能使用 java.util.ArrayList 类型)[重复 1]

rdd.saveAsNewAPIHadoopFile( 
            path='-', 
            outputFormatClass="org.elasticsearch.hadoop.mr.EsOutputFormat", 
            keyClass="org.apache.hadoop.io.NullWritable", 
            valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable", 
            conf={ 
        "es.nodes" : "localhost", 
        "es.port" : "9200", 
        "es.resource" : "mboyd/mboydtype" 
    }) 

我希望文档在写入 ES 时最终的样子是:

{
field:[{"key1":"val1"}, {"key2":"val2"}]
}

Answer 1

游戏有点晚了，但这是我们在昨天遇到这个问题后想出的解决方案。将 'es.input.json': 'true' 添加到您的 conf，然后对您的数据运行 json.dumps()。

修改您的示例，如下所示:

import json

rdd = sc.parallelize([{"key1": ["val1", "val2"]}])
json_rdd = rdd.map(json.dumps)
json_rdd.saveAsNewAPIHadoopFile( 
    path='-', 
    outputFormatClass="org.elasticsearch.hadoop.mr.EsOutputFormat", 
    keyClass="org.apache.hadoop.io.NullWritable", 
    valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable", 
    conf={ 
        "es.nodes" : "localhost", 
        "es.port" : "9200", 
        "es.resource" : "mboyd/mboydtype",
        "es.input.json": "true"
    }
)