从 ES 加载数据并使用 pig 在 HDFS 中存储为 avro

Posted 2023-04-18

【中文标题】从 ES 加载数据并使用 pig 在 HDFS 中存储为 avro

【英文标题】：Load data from ES and store as avro in HDFS using pig

【发布时间】：2017-04-27 09:45:34

【问题描述】：

我有一些关于 ElasticSearch 的数据需要在 HDFS 上发送。我正在尝试使用 pig（这是我第一次使用它），但是在为我的数据定义正确的架构时遇到了一些问题。

首先，我尝试使用'es.output.json=true' 和org.elasticsearch.hadoop.pig.EsStorage 选项加载JSON，我可以正确加载/转储数据，还可以使用STORE A INTO 'hdfs://path/to/store'; 将它们作为JSON 保存到HDFS。稍后，在 HIVE 上定义一个外部表，我可以查询这些数据。这是运行良好的完整示例（我从代码中删除了所有 SSL 属性）：

REGISTER /path/to/commons-httpclient-3.1.jar;
REGISTER /path/to/elasticsearch-hadoop-5.3.0.jar;

A = LOAD 'my-index/log' USING org.elasticsearch.hadoop.pig.EsStorage(
'es.nodes=https://addr1:port,https://addr2:port2,https://addr3:port3',
'es.query=?q=*',
'es.output.json=true');

STORE A INTO 'hdfs://path/to/store';

如何将我的数据作为 AVRO 存储到 HDFS？我想我需要使用AvroStorage，但我还应该定义一个加载数据的模式，或者JSON就足够了？我尝试使用LOAD...USING...AS 命令定义架构并设置es.mapping.date.rich=false 而不是es.output.json=true（我的数据非常复杂，带有地图之类的地图），但它不起作用。我不确定问题出在语法上，还是出在方法本身上。很高兴能提示正确的方向。

更新

这是我尝试使用es.mapping.date.rich=false 的示例。我的问题是，如果一个字段为空，所有字段的顺序都会错误。

A = LOAD 'my-index/log' USING org.elasticsearch.hadoop.pig.EsStorage(
  'es.nodes=https://addr1:port,https://addr2:port2,https://addr3:port3',
  'es.query=?q=*',
  'es.mapping.date.rich=false')
  AS(
    field1:chararray,
    field2:chararray,
    field3:map[chararray,fieldMap:map[],chararray],
    field4:chararray,
    field5:map[]
  );

B = FOREACH A GENERATE field1, field2;

STORE B INTO 'hdfs://path/to/store' USING AvroStorage('

  "type" : "foo1",
  "name" : "foo2",
  "namespace" : "foo3",
  "fields" : [ 
    "name" : "field1",
    "type" : ["null","string"],
    "default" : null
  , 
    "name" : "field2",
    "type" : ["null","string"],
    "default" : null
   ]

');

【参考方案1】：

对于未来的读者，我决定改用spark，因为它比pig 快得多。要将avro 文件保存在hdfs 上，我使用的是databrick 库。