這里以將Apache的日志寫(xiě)入到ElasticSearch為例，來(lái)演示一下如何使用Python將Spark數(shù)據(jù)導(dǎo)入到ES中。

實(shí)際工作中，由于數(shù)據(jù)與使用框架或技術(shù)的復(fù)雜性，數(shù)據(jù)的寫(xiě)入變得比較復(fù)雜，在這里我們簡(jiǎn)單演示一下。

如果使用Scala或Java的話(huà)，Spark提供自帶了支持寫(xiě)入ES的支持庫(kù)，但Python不支持。所以首先你需要去這里下載依賴(lài)的ES官方開(kāi)發(fā)的依賴(lài)包包。

下載完成后，放在本地目錄，以下面命令方式啟動(dòng)pyspark:

pyspark --jars elasticsearch-hadoop-6.4.1.jar

如果你想pyspark使用Python3，請(qǐng)?jiān)O(shè)置環(huán)境變量：

export PYSPARK_PYTHON=/usr/bin/python3理解如何寫(xiě)入ES的關(guān)鍵是要明白，ES是一個(gè)JSON格式的數(shù)據(jù)庫(kù)，它有一個(gè)必須的要求。數(shù)據(jù)格式必須采用以下格式

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

往下會(huì)展示如何轉(zhuǎn)換成這種格式。

解析Apache日志文件我們將Apache的日志文件讀入，構(gòu)建Spark RDD。然后我們寫(xiě)一個(gè)parse()函數(shù)用正則表達(dá)式處理每條日志，提取我們需要的字

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

換句話(huà)說(shuō)，我們剛開(kāi)始從日志文件讀入RDD的數(shù)據(jù)類(lèi)似如下：

[’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函數(shù)轉(zhuǎn)換每條記錄：

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’}]

現(xiàn)在看起來(lái)像JSON，但并不是JSON字符串，我們需要使用json.dumps將dict對(duì)象轉(zhuǎn)換。

我們同時(shí)增加一個(gè)doc_id字段作為整個(gè)JSON的ID。在配置ES中我們?cè)黾尤缦屡渲谩癳s.mapping.id”: “doc_id”告訴ES我們將這個(gè)字段作為ID。

這里我們使用SHA算法，將這個(gè)JSON字符串作為參數(shù)，得到一個(gè)唯一ID。計(jì)算結(jié)果類(lèi)似如下，可以看到ID是一個(gè)很長(zhǎng)的SHA數(shù)值。

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'}’)]

現(xiàn)在我們需要制定ES配置，比較重要的兩項(xiàng)是：

“es.resource” : ‘walker/apache’: 'walker'是索引，apache是類(lèi)型，兩者一般合稱(chēng)索引 “es.mapping.id”: “doc_id”： 告訴ES那個(gè)字段作為整個(gè)文檔的ID，也就是查詢(xún)結(jié)果中的_id

其他的配置自己去探索。

然后我們使用saveAsNewAPIHadoopFile()將RDD寫(xiě)入到ES。這部分代碼對(duì)于所有的ES都是一樣的，比較固定，不需要理解每一個(gè)細(xì)節(jié)

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進(jìn)行查詢(xún)

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 jsonimport hashlibimport redef 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)

也可以這么封裝，其實(shí)原理是一樣的

import hashlibimport jsonfrom pyspark import Sparkcontextdef 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 #記得這邊返回的是字典類(lèi)型的，在寫(xiě)入es之前要記得dumpsdef saveData2es(pdd, es_host, port,index, index_type, key): ''' 把saprk的運(yùn)行結(jié)果寫(xiě)入es :param pdd: 一個(gè)rdd類(lèi)型的數(shù)據(jù) :param es_host: 要寫(xiě)es的ip :param index: 要寫(xiě)入數(shù)據(jù)的索引 :param index_type: 索引的類(lèi)型 :param key: 指定文檔的id，就是要以文檔的那個(gè)字段作為_(kāi)id :return: ''' #實(shí)例es客戶(hù)端記得單例模式 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)))) #這百年是為把這個(gè)數(shù)據(jù)構(gòu)造成元組格式，如果傳進(jìn)來(lái)的_dic是字典則需要jdumps，如果傳進(jìn)來(lái)之前就已經(jīng)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__’: #實(shí)例化sp對(duì)象 sc=Sparkcontext() #文件中的呢內(nèi)容一行一行用sc的讀取出來(lái) json_text=sc.textFile(’./1.txt’) #進(jìn)行轉(zhuǎn)換 json_data=json_text.map(lambda line:parse(line)) saveData2es(json_data,’127.0.01’,’9200’,’index_test’,’index_type’,’doc_id’) sc.stop()

看到了把，面那個(gè)例子在寫(xiě)入es之前加了一個(gè)id，返回一個(gè)元組格式的，現(xiàn)在這個(gè)封裝指定_id就會(huì)比較靈活了

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持好吧啦網(wǎng)。