Python如何把Spark数据写入ElasticSearch

p=re.compile(regex)
def parse(str):
  s=p.match(str)
  d = {}
  d['ip']=s.group(1)
  d['date']=s.group(4)
  d['operation']=s.group(5)
  d['uri']=s.group(6)
  return d

es_write_conf = {
    "es.nodes" : "localhost",
    "es.port" : "9200",
    "es.resource" : 'walker/apache',
    "es.input.json": "yes",
    "es.mapping.id": "doc_id"
  }
    
rdd3.saveAsNewAPIHadoopFile(
    path='-',
   outputFormatClass="org.elasticsearch.hadoop.mr.EsOutputFormat",    keyClass="org.apache.hadoop.io.NullWritable",
    valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable",
    conf=es_write_conf)

rdd3 = rdd2.map(addID)

def addId(data):
  j=json.dumps(data).encode('ascii', 'ignore')
  data['doc_id'] = hashlib.sha224(j).hexdigest()
  return (data['doc_id'], json.dumps(data))

curl http://localhost:9200s/walker/apache/_search?pretty=true&?q=*
{
    "_index" : "walker",
    "_type" : "apache",
    "_id" : "227e977849bfd5f8d1fca69b04f7a766560745c6cb3712c106d590c2",
    "_score" : 1.0,
    "_source" : {
     "date" : "17/May/2015:10:05:32 +0000",
     "ip" : "91.177.205.119",
     "operation" : "GET",
     "doc_id" : "227e977849bfd5f8d1fca69b04f7a766560745c6cb3712c106d590c2",
     "uri" : "/favicon.ico"
    }

import json
import hashlib
import re

def addId(data):
  j=json.dumps(data).encode('ascii', 'ignore')
  data['doc_id'] = hashlib.sha224(j).hexdigest()
  return (data['doc_id'], json.dumps(data))

def parse(str):
  s=p.match(str)
  d = {}
  d['ip']=s.group(1)
  d['date']=s.group(4)
  d['operation']=s.group(5)
  d['uri']=s.group(6)
  return d  

regex='^(\S+) (\S+) (\S+) \[([\w:/]+\s[+\-]\d{4})\] "(\S+)\s?(\S+)?\s?(\S+)?" (\d{3}|-) (\d+|-)\s?"?([^"]*)"?\s?"?([^"]*)?"?$'

p=re.compile(regex)

rdd = sc.textFile("/home/ubuntu/walker/apache_logs")

rdd2 = rdd.map(parse)

rdd3 = rdd2.map(addID)

es_write_conf = {
    "es.nodes" : "localhost",
    "es.port" : "9200",
    "es.resource" : 'walker/apache',
    "es.input.json": "yes",
    "es.mapping.id": "doc_id"
  }
   
rdd3.saveAsNewAPIHadoopFile(
    path='-',
   outputFormatClass="org.elasticsearch.hadoop.mr.EsOutputFormat",    keyClass="org.apache.hadoop.io.NullWritable",
    valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable",
    conf=es_write_conf)

import hashlib
import json
from pyspark import Sparkcontext

def make_md5(line):
  md5_obj=hashlib.md5()
  md5_obj.encode(line)
  return md5_obj.hexdigest()

def parse(line):
  dic={}
  l = line.split('\t')
  doc_id=make_md5(line)
  dic['name']=l[1]
  dic['age'] =l[2]
  dic['doc_id']=doc_id
  return dic  #记得这边返回的是字典类型的，在写入es之前要记得dumps

def saveData2es(pdd, es_host, port,index, index_type, key):
  """
  把saprk的运行结果写入es
  :param pdd: 一个rdd类型的数据
  :param es_host: 要写es的ip
  :param index: 要写入数据的索引
  :param index_type: 索引的类型
  :param key: 指定文档的id，就是要以文档的那个字段作为_id
  :return:
  """
  #实例es客户端记得单例模式
  if es.exist.index(index):
    es.index.create(index, 'spo')
  es_write_conf = {
    "es.nodes": es_host,
    "es.port": port,
    "es.resource": index/index_type,
    "es.input.json": "yes",
    "es.mapping.id": key
  }

  (pdd.map(lambda _dic: ('', json.dumps(_dic))))  #这百年是为把这个数据构造成元组格式，如果传进来的_dic是字典则需要jdumps，如果传进来之前就已经dumps，这便就不需要dumps了
  .saveAsNewAPIHadoopFile(
    path='-',
    outputFormatClass="org.elasticsearch.hadoop.mr.EsOutputFormat", keyClass="org.apache.hadoop.io.NullWritable",
    valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable",
    conf=es_write_conf)
  )
if __name__ == '__main__':
  #实例化sp对象
  sc=Sparkcontext()
  #文件中的呢内容一行一行用sc的读取出来
  json_text=sc.textFile('./1.txt')
  #进行转换
  json_data=json_text.map(lambda line:parse(line))

  saveData2es(json_data,'127.0.01','9200','index_test','index_type','doc_id')

  sc.stop()