如何将 Spark Dataframe 推送到 Elastic Search (Pyspark)

Question

这里是初级 ES 问题

将 Spark Dataframe 推送到 Elastic Search 的工作流程或步骤是什么？

根据研究，我相信我需要使用spark.newAPIHadoopFile() 方法。

但是，在挖掘 Elastic Search Documentation 和 other Stack Q/A's 时，我仍然对参数需要采用什么格式以及为什么感到有些困惑

请注意，我使用的是 pyspark，这是 ES 的新表（不存在索引），df 是 5 列（2 个字符串类型、2 个长类型和 1 个整数列表），行数约为 350 万.

Answer 1

设法找到答案，所以我会分享。 Spark DF（来自 pyspark.sql）目前不支持 newAPIHadoopFile() 方法；但是，df.rdd.saveAsNewAPIHadoopFile() 也给了我错误。诀窍是通过以下函数将 df 转换为字符串

def transform(doc):
    import json
    import hashlib

    _json = json.dumps(doc)
    keys = doc.keys()
    for key in keys:
        if doc[key] == 'null' or doc[key] == 'None':
            del doc[key]
    if not doc.has_key('id'):
        id = hashlib.sha224(_json).hexdigest()
        doc['id'] = id
    else:
        id = doc['id']
    _json = json.dumps(doc)
    return (id, _json)

所以我的 JSON 工作流程是：

1：df = spark.read.json('XXX.json')

2：rdd_mapped = df.rdd.map(lambda y: y.asDict())

3：final_rdd = rdd_mapped.map(transform)

4：

final_rdd.saveAsNewAPIHadoopFile(
     path='-', 
     outputFormatClass="org.elasticsearch.hadoop.mr.EsOutputFormat",
     keyClass="org.apache.hadoop.io.NullWritable",  
     valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable", 
     conf={ "es.resource" : "<INDEX> / <INDEX>", "es.mapping.id":"id", 
         "es.input.json": "true", "es.net.http.auth.user":"elastic",
         "es.write.operation":"index", "es.nodes.wan.only":"false",
         "es.net.http.auth.pass":"changeme", "es.nodes":"<NODE1>, <NODE2>, <NODE3>...",
         "es.port":"9200" })

更多information on ES arguments can be found here（滚动到“配置”）

Answer 2

这对我有用 - 我的数据在 df。

df = df.drop('_id')
df.write.format(
    "org.elasticsearch.spark.sql"
).option(
    "es.resource", '%s/%s' % (conf['index'], conf['doc_type'])
).option(
    "es.nodes", conf['host']
).option(
    "es.port", conf['port']
).save()

我曾使用此命令提交我的作业 - /path/to/spark-submit --master spark://master:7077 --jars ./jar_files/elasticsearch-hadoop-5.6.4.jar --driver-class-path ./jar_files/elasticsearch-hadoop-5.6.4.jar main_df.py。