0

我在将数据从 spark 流 (pyspark) 索引到 elasticserach 时遇到问题。数据是类型dstream。下面是它的外观

(u'01B', 0)
(u'1A5', 1)
....

这是我正在使用的弹性索引：index=clus 和 type=data

GET /clus/_mapping/data
{
   "clus": {
      "mappings": {
         "data": {
            "properties": {
               "content": {
                  "type": "text"
               }
            }
         }
      }
   }
}

这是我的代码：

ES_HOST = {
    "host" : "localhost", 
    "port" : 9200
}

INDEX_NAME = 'clus'
TYPE_NAME = 'data'
ID_FIELD = 'responseID' 

# create ES client
es = Elasticsearch(hosts = [ES_HOST])

# some config before sending to elastic     
if not es.indices.exists(INDEX_NAME):
    request_body = {
        "settings" : {
            "number_of_shards": 1,
            "number_of_replicas": 0
        }
    }
    res = es.indices.create(index = INDEX_NAME, body = request_body)
es_write_conf = {
        "es.nodes": "localhost",
        "es.port": "9200",
        "es.resource": INDEX_NAME+"/"+TYPE_NAME
    }
sc = SparkContext(appName="PythonStreamingKafka")
    ssc = StreamingContext(sc, 30)

# .....
#loading data to put in elastic : lines4

    lines4.foreachRDD(lambda rdd: rdd.saveAsNewAPIHadoopFile(
        path='-',
        outputFormatClass="org.elasticsearch.hadoop.mr.EsOutputFormat",
        keyClass="org.apache.hadoop.io.NullWritable",
        valueClass="org.elasticsearch.hadoop.mr.LinkedMapWritable",
        conf=es_write_conf))




    ssc.start()
    ssc.awaitTermination()

这是错误：

17/07/25 15:31:31 错误执行程序：阶段 11.0（TID 23）中任务 2.0 中的异常 org.elasticsearch.hadoop.rest.EsHadoopInvalidRequest：发现不可恢复的错误 [127.0.0.1:9200] 返回错误请求（400） - 解析失败；在 org.elasticsearch.hadoop.rest.RestClient.bulk(RestClient.java:203) 在 org.elasticsearch.hadoop.rest 的 org.elasticsearch.hadoop.rest.RestClient.processBulkResponse(RestClient.java:251) .RestRepository.tryFlush(RestRepository.java:220) 在 org.elasticsearch.hadoop.rest.RestRepository.flush(RestRepository.java:242) 在 org.elasticsearch.hadoop.rest.RestRepository.close(RestRepository.java:267) 在org.elasticsearch.hadoop.mr.EsOutputFormat$EsRecordWriter.doClose(EsOutputFormat.java:214) 在 org.elasticsearch.hadoop.mr.EsOutputFormat$EsRecordWriter.close(EsOutputFormat.java: ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:748) 17/07/25 15:31:31 错误执行器：阶段 11.0 中的任务 0.0 异常（TID 21 ) org.elasticsearch.hadoop.rest.EsHadoopInvalidRequest：发现不可恢复的错误 [127.0.0.1:9200] 返回错误请求（400） - 解析失败；在 org.elasticsearch.hadoop.rest.RestClient.bulk(RestClient.java:203) 在 org.elasticsearch.hadoop.rest 的 org.elasticsearch.hadoop.rest.RestClient.processBulkResponse(RestClient.java:251) .RestRepository.tryFlush(RestRepository.java:220) 在 org.elasticsearch.hadoop.rest.RestRepository.flush(RestRepository.java:242) 在 org.elasticsearch.hadoop.rest.RestRepository.close(RestRepository.java:267) 在org.elasticsearch.hadoop.mr.EsOutputFormat$EsRecordWriter.doClose(EsOutputFormat. ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 在 java.lang.Thread.run(Thread.java:748) 17/07/25 15 :31:31 错误执行程序：阶段 11.0（TID 22）中任务 1.0 中的异常 org.elasticsearch.hadoop.rest.EsHadoopInvalidRequest：发现不可恢复的错误 [127.0.0.1:9200] 返回错误请求（400） - 无法解析；在 org.elasticsearch.hadoop.rest.RestClient.bulk(RestClient.java:203) 在 org.elasticsearch.hadoop.rest 的 org.elasticsearch.hadoop.rest.RestClient.processBulkResponse(RestClient.java:251) .RestRepository.tryFlush(RestRepository.java:220) 在 org.elasticsearch.hadoop.rest.RestRepository.flush(RestRepository.java:242) 在 org.elasticsearch.hadoop.rest.RestRepository.close(RestRepository.java:

4

1 回答 1

1

您创建索引的方式似乎存在错误。您需要在创建索引时发送mapping请求body。这是一个工作示例：

from elasticsearch import Elasticsearch 

es = Elasticsearch(["http://localhost:9200"])
# create index 
index_name = "clus" 
index_mapping = {
   "clus": {
      "mappings": {
         "data": {
            "properties": {
               "content": {
                  "type": "text"
               }
            }
         }
      }
   }
}


if not es.indices.exists(index_name):
        res = es.indices.create(index=index_name, body=index_mapping)
        print res

您应该将其{u'acknowledged': True}作为确认您的索引已创建的响应。

然后，您使用 foreachRDD 遍历您的数据 dstream，并应用一个将数据转换为 json 结构{"content": str((u'1A5', 1))}并对其进行索引的函数，如下所示

doc = {"content": str((u'1A5', 1))}
res = es.index(index="clus", doc_type='data', body=doc)

附带说明一下，不建议将数据索引为列表(u'1A5', 1)，这样您将很难在其他上下文中使用它，例如在 kibana 上进行可视化。

于 2017-07-25T15:50:20.517 回答