使用配置单元/Qubole插入到ElasticSearch

Question

我正在尝试将数据从配置单元表格插入到elastic search中。

    CREATE EXTERNAL TABLE IF NOT EXISTS es_temp_table (
        dt    STRING,
        text  STRING
    )
    STORED BY 'org.elasticsearch.hadoop.hive.EsStorageHandler'
    TBLPROPERTIES('es.resource'='aggr_2014-10-01/metric','es.index.auto.create'='true')
    ;

    INSERT OVERWRITE TABLE es_temp_table
    SELECT dt, description
    FROM other_table

然而，数据是关闭的。当我在另一个表上执行count(*)时，我得到了6000行。当我搜索aggr_2014-10-01索引时，我看到了10,000条记录！不知何故，记录被复制了(行被复制了多次)。也许我可以删除elastic search中的重复记录？不过，我不确定我该怎么做。

如果一个映射器成功，它会试图杀死另一个映射器。然而，另一个任务已经造成了破坏(也就是插入到ElasticSearch中)。

set mapred.map.tasks.speculative.execution=false;

我发现的一件事是将推测性执行设置为false，这样每个映射器只产生一个任务(参见上面的设置)。然而，现在我看到了被低估的情况。我相信这可能是因为记录被跳过了，但我无法诊断为什么这些记录会被跳过。

在这个版本中，这也意味着即使一个任务/映射器失败，整个作业也会失败，然后我需要删除索引(上传了部分数据)并重新运行整个作业(这需要大约4个小时)。

进度更新

我试图通过将所有工作放在reducer中来解决这个问题(这是唯一一种只产生一个任务以确保不会插入重复记录的方法)。

INSERT OVERWRITE TABLE es_temp_table
SELECT dt, description
FROM other_table
DISTRIBUTE BY cast(rand()*250 as int);

然而，我现在看到了一个巨大的低估！现在只有2000条记录。Elastic search确实估计了一些事情，但不是到了这个程度。ElasticSearch中只有很少的记录。这可能是由于失败的任务(不再重试)造成的。它可能来自Qubole/Hive传递格式错误的条目。但我设置了：

set mapreduce.map.skip.maxrecords=1000;

以下是我的查询的其他一些设置：

set es.nodes=node-names
set es.port=9200;
set es.bulk.size.bytes=1000mb;
set es.http.timeout=20m;
set mapred.tasktracker.expiry.interval=3600000;
set mapred.task.timeout=3600000;

Answer 1

我确定了问题所在。正如我所怀疑的那样，插入跳过了一些被认为是“坏”的记录。这就解决了问题！不再跳过记录，所有数据都将上载到Elastic Search。

INSERT OVERWRITE TABLE es_temp_table
SELECT dt, REGEXP_REPLACE(description, '[^0-9a-zA-Z]+', ' ')
FROM other_table