问Spark在写入elasticsearch时不支持arraylist吗？

EN

这个游戏有点晚了，但这是我们昨天遇到这个问题后提出的解决方案。将'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"
    }
) 

刚刚遇到了这个问题，解决方案通过将所有列表转换为元组来解决。转换为json也是如此。

我觉得在其他答案中有几点缺失，比如你必须从你的RDD返回一个2元组(我不知道为什么)，并且还需要Elasticsearch hadoop jar文件来使其工作。因此，我将编写我必须遵循的整个过程，以使其工作。

1. 下载Elasticsearch Hadoop jar文件。您可以从central maven repository下载它(最新版本在大多数情况下都可以使用-查看他们的official requirements README了解更多信息)。
2. 使用以下最少的代码片段为演示创建一个run.py文件。

import json import pymongo_spark pymongo_spark.activate() from pyspark import SparkContext，SparkConf conf = SparkConf().setAppName('demo').setMaster('local') sc = SparkContext(conf=conf) rdd = sc.parallelize([{"key1"："val1"，"val2"}]) final_rdd = rdd.map(json.dumps).map(lambda x：('key'，x)) final_rdd.saveAsNewAPIHadoopFile( path='-'，pathkeyClass="org.apache.hadoop.io.NullWritable"，valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable"，conf={ "es.nodes“：""，"es.port”："9200"，"es.resource“：”索引名称/文档类型名称“，"es.input.json"："true“})使用以下命令./bin/spark-submit --jars /path/to/your/jar/file/elasticsearch-hadoop-5.6.4.jar --driver-class-path /path/to/you/jar/file/elasticsearch-hadoop-5.6.4.jar --master yarn /path/to/your/run/file/run.py运行

• 作业

哈！