如何使用python将Spark数据写入ElasticSearch

Posted 阿布alone

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了如何使用python将Spark数据写入ElasticSearch相关的知识,希望对你有一定的参考价值。

这里以将Apache的日志写入到ElasticSearch为例,来演示一下如何使用Python将Spark数据导入到ES中。

实际工作中,由于数据与使用框架或技术的复杂性,数据的写入变得比较复杂,在这里我们简单演示一下。

如果使用Scala或Java的话,Spark提供自带了支持写入ES的支持库,但Python不支持。所以首先你需要去这里下载依赖的ES官方开发的依赖包包。

下载完成后,放在本地目录,以下面命令方式启动pyspark:

     pyspark --jars elasticsearch-hadoop-6.4.1.jar

 

如果你想pyspark使用Python3,请设置环境变量:

 export PYSPARK_PYTHON=/usr/bin/python3

理解如何写入ES的关键是要明白,ES是一个JSON格式的数据库,它有一个必须的要求。数据格式必须采用以下格式

 
{ "id: { the rest of your json}}

 

往下会展示如何转换成这种格式。

解析Apache日志文件

我们将Apache的日志文件读入,构建Spark RDD。然后我们写一个parse()函数用正则表达式处理每条日志,提取我们需要的字

rdd = sc.textFile("/home/ubuntu/walker/apache_logs")
regex=^(S+) (S+) (S+) [([w:/]+s[+-]d{4})] "(S+)s?(S+)?s?(S+)?" (d{3}|-) (d+|-)s?"?([^"]*)"?s?"?([^"]*)?"?$
 
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  

 

换句话说,我们刚开始从日志文件读入RDD的数据类似如下:

[83.149.9.216 - - [17/May/2015:10:05:03 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1" 200 203023 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"]

 

然后我们使用map函数转换每条记录:

rdd2 = rdd.map(parse)

rdd2.take(1)

[{date: 17/May/2015:10:05:03 +0000, ip: 83.149.9.216, operation: GET, uri: /presentations/logstash-monitorama-2013/images/kibana-search.png}]

 

现在看起来像JSON,但并不是JSON字符串,我们需要使用json.dumps将dict对象转换。

我们同时增加一个doc_id字段作为整个JSON的ID。在配置ES中我们增加如下配置“es.mapping.id”: “doc_id”告诉ES我们将这个字段作为ID。

这里我们使用SHA算法,将这个JSON字符串作为参数,得到一个唯一ID。
计算结果类似如下,可以看到ID是一个很长的SHA数值。

rdd3.take(1)

[(a5b086b04e1cc45fb4a19e2a641bf99ea3a378599ef62ba12563b75c, {"date": "17/May/2015:10:05:03 +0000", "ip": "83.149.9.216", "operation": "GET", "doc_id": "a5b086b04e1cc45fb4a19e2a641bf99ea3a378599ef62ba12563b75c", "uri": "/presentations/logstash-monitorama-2013/images/kibana-search.png"})]

 

 

现在我们需要制定ES配置,比较重要的两项是:

  • “es.resource” : ‘walker/apache’: "walker"是索引,apache是类型,两者一般合称索引
  • “es.mapping.id”: “doc_id”: 告诉ES那个字段作为整个文档的ID,也就是查询结果中的_id
    其他的配置自己去探索。

然后我们使用saveAsNewAPIHadoopFile()将RDD写入到ES。这部分代码对于所有的ES都是一样的,比较固定,不需要理解每一个细节

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进行查询

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)
        

 




链接:https://www.jianshu.com/p/c7365b9bda0a

以上是关于如何使用python将Spark数据写入ElasticSearch的主要内容,如果未能解决你的问题,请参考以下文章

如何将 Spark 数据帧写入 impala 数据库

使用 spark 写入 parquet 文件时如何添加额外的元数据

如何将 Spark 结构化流数据写入 Hive?

Spark - 如何将约 20TB 的数据从 DataFrame 写入配置单元表或 hdfs?

如何在 Spark SQL 中使用 snappy 压缩将数据写入配置单元表

如何并行处理数据但将结果写入 Spark 中的单个文件