如何将Spark数据框推送到elasticsearch（Pyspark）

如何将Spark数据框推送到elasticsearch（Pyspark）

设法找到答案，所以我会分享。SparkDF（来自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" })

有关ES参数的更多信息，请参见此处（滚动到“配置”）