0

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

首先，我尝试使用 选项加载 JSON 'es.output.json=true'，org.elasticsearch.hadoop.pig.EsStorage我可以正确加载/转储数据，还可以使用STORE A INTO 'hdfs://path/to/store';. 稍后，在 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
  } ]
}
');
4

1 回答 1

0

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

于 2017-08-02T13:07:08.493 回答