Firehose 记录格式转换分区答案

【问题标题】：Firehose record format conversion partitionsFirehose 记录格式转换分区
【发布时间】：2018-10-24 21:57:32
【问题描述】：

我尝试使用新的 firehose 功能“记录格式转换”将我的事件保存为 parquet 文件以用于 athena 或 hive 聚合。您必须从胶水目录中选择表，但 firehose 会忽略已定义的分区，而是将文件保存在 YYYY/MM/DD/HH/ 结构中。数据也缺少定义的分区列。如果它用于分区，那就没问题了。

是否有 API 配置或其他东西强制使用表分区？

【问题讨论】：

标签： file-conversion amazon-kinesis-firehose

【解决方案1】：

我有完全相同的问题，即使使用相同的分区

所以你必须使用 AWS lambda 来实现你想要的

用于将 Firehose 生成的文件移动到 Athena 使用的存储桶中。
另一个触发刷新 Athena 表，因为它不会看到新文件夹（我没有设置所有触发器，但这应该只是一个调用 'MSCK REPAIR TABLE your_table_name;'）

对于第一个我选择 NodeJs，因为它非常简单而且非常快。 ~ 3 秒移动 120MB 文件，最小 AWS 允许 128MB RAM 内存分配（Firehose 生成的文件最大约为 64MB）

Node js项目结构包.json

{ 
  "name": "your.project", 
  "version": "1.0.0", 
  "description": "Copy generated partitioned files by Firehose to valid partitioned files for Athena", 
  "main": "index.js", 
  "dependencies": { 
    "async": "^2.6.1" 
   } 
}

还有 index.js

const aws = require('aws-sdk');
const async = require('async');
const s3 = new aws.S3();
const dstBucket = 'PUT_YOUR_BUCKET_NAME_HERE';
var util = require('util');

exports.handler = (event, context, callback) => {
    const srcBucket = event.Records[0].s3.bucket.name;
    const srcKey = event.Records[0].s3.object.key;
    const split = srcKey.split('/');
    const dstKey = `event_year=${split[0]}/event_month=${split[1]}/event_day=${split[2]}/event_hour=${split[3]}/${split[4]}`;
    console.log("Reading options from event:\n", util.inspect(event, {depth: 10}));
    async.waterfall([
            function copy(next) {
                s3.copyObject({
                    Bucket: dstBucket,
                    CopySource: `${srcBucket}/${srcKey}`,
                    Key: dstKey
                }, next);
            },
            function deleteOriginal(copyResult, next) {
                s3.deleteObject({
                    Bucket: srcBucket,
                    Key: srcKey
                }, next);
            }
        ], function (err) {
            if (err) {
                console.error(`Failed: ${srcBucket}/${srcKey} => ${dstBucket}/${dstKey} to move FireHose partitioned object to Athena partitioned object. Error: ${err}`);
            } else {
                console.log(`Success: ${srcBucket}/${srcKey} => ${dstBucket}/${dstKey} moved FireHose partitioned object to Athena partitioned object`);
            }
            callback(null, 'move success');
        }
    );
};

只需更新一些数据以对您的案例有效。我遇到的另一个问题是在构建项目时使用

npm install

然后压缩它，它在 AWS 解压缩中是不正确的，所以我必须更新我的 index.js 的路径。

这行得通。

你也可以找到这一行

console.log("Reading options from event:\n", util.inspect(event, {depth: 10}));

可以去掉，但对理解加工对象的细节有很大帮助

【讨论】：