Elasticsearch parallel_bulk上的连接超时

Question

我正在尝试使用Elasticsearchparallel_bulk导入大量数据。这是我的索引结构：

{
    "_index" : "myindex",
    "_type" : domain,
    "_id" : md5(email),
    "_score" : 1.0,
    "_source" : {
      "purchase_date" : purchase_date,
      "amount" : amount,
    }
}

这是我的python代码：

def insert(input_file):
    paramL = []

    with open(input_file) as f:
        for line in f:
            line = line.rstrip()

            fields = line.split(',')
            purchase_date = fields[0]
            amount = fields[1]
            email = fields[2]               

            id_email = getMD5(email)

            doc = {
                "email": email,
                "purchase_date": purchase_date,
                "amount": amount _date
            }

            ogg = {
                '_op_type': 'index',
                '_index': index_param,
                '_type': doctype_param,
                '_id': id_email,
                '_source': doc
            }

            paramL.append(ogg)    

            if len(paramL) > 500000:
                for success, info in helpers.parallel_bulk(client=es, actions=paramL, thread_count=4):
                    if not success:
                        print "Insert failed: ", info

                # empty paramL if size > 5.000.000
                del paramL[:]

该文件包含42.644.394行，我认为每次列表“paramL”大约是5.000.000个元素时都要插入数据。因此，当我运行该脚本时，它插入了大约436.226个值，直到它崩溃并出现以下错误:

回溯（最近一次调用）： 文件 “test-2-0.py”，第 133 行，在 main（） 文件 “test-2-0.py”，第 131 行，在 main insert（args.file） 文件 “test-2-0.py”，第 82 行，在插入中表示成功，在 helpers.parallel_bulk 中的信息（client=es，actions=paramL，thread_count=4）：文件 “/usr/local/lib/python2.7/dist-packages/elasticsearch/helpers/init.py”，第 306 行，在 parallel_bulk _chunk_actions 中（actions， chunk_size， max_chunk_bytes， client.transport.serializer） 文件 “/usr/lib/python2.7/multiprocessing/pool.py”，第 668 行，在下一个提升值 elasticsearch.exceptions.ConnectionTimeout： ConnectionTimeout 由 - ReadTimeoutError（HTTPConnectionPool（host=u'127.0.0.1'， port=9200） 引起：读取超时。（读取超时 = 10））

我还试图增加超时传递它在Elasticsearch构造函数

es = Elasticsearch(['127.0.0.1'], request_timeout=30)

但结果是一样的。

Answer 1

诚恳地说，我从来没有批量导入这么多要标记的文档。我不知道为什么会出现这个错误。在您的情况下，我建议不要创建列表-paramL，而是使用生成器函数来管理您的数据，正如弹性开发人员在弹性论坛中描述的大批量摄取的最佳实践：https://discuss.elastic.co/t/helpers-parallel-bulk-in-python-not-working/39498/3。大致如下：

def insert(input_file):

    with open(input_file) as f:
        for line in f:
            line = line.rstrip()

            fields = line.split(',')
            purchase_date = fields[0]
            amount = fields[1]
            email = fields[2]               

            id_email = getMD5(email)

            doc = {
                "email": email,
                "purchase_attack": purchase_date,
                "amount _relevation": amount _date
            }

            yield {
                '_op_type': 'index',
                '_index': index_param,
                '_type': doctype_param,
                '_id': id_email,
                '_source': doc
            }



for success, info in helpers.parallel_bulk(client=es, actions=insert(input_file), thread_count=4):
    if not success:
        print "Insert failed: ", info

您可以在java虚拟机中增加专用于elastic的空间编辑此文件< code >/etc/elastic search/JVM . options 要分配2 GB的RAM，您应该更改——如果您的机器有4 GB，您应该为系统保留近1 GB，因此您最多可以分配3 gb:

# Xms represents the initial size of total heap space
# Xmx represents the maximum size of total heap space

 -Xms2g
 -Xmx2g

然后，您必须重新启动服务

sudo service elasticsearch restart

再试一次。祝你好运