AWS Java CDK Alpha版：Kinesis Firehose转Parquet格式配置问询

核心思路 #

由于Alpha版本CDK暂未提供S3Bucket.processor的直接实现，可通过直接构造Firehose底层API配置的方式，手动添加格式转换与Glue表关联参数，绕开高层封装限制。

具体实现步骤 #

1. 提前创建Glue数据库与Parquet格式表 #

确保Glue表结构与Firehose接收的数据结构匹配，示例代码：

GlueDatabase database = new GlueDatabase(this, "FirehoseParquetDb", GlueDatabaseProps.builder()
        .databaseName("firehose_parquet_db")
        .build());

GlueTable table = new GlueTable(this, "FirehoseParquetTable", GlueTableProps.builder()
        .database(database)
        .tableName("firehose_parquet_table")
        .storageDescriptor(StorageDescriptor.builder()
                .columns(List.of(
                        Column.builder().name("id").type("int").build(),
                        Column.builder().name("event_time").type("timestamp").build(),
                        Column.builder().name("payload").type("string").build()
                ))
                .location("s3://" + s3Bucket.getBucketName() + "/data/")
                .inputFormat("org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat")
                .outputFormat("org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat")
                .serdeInfo(SerdeInfo.builder()
                        .serializationLibrary("org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe")
                        .build())
                .build())
        .partitionKeys(List.of(Column.builder().name("year").type("string").build()))
        .build());

2. 手动构造Firehose S3目标的格式转换配置 #

通过CfnDeliveryStream的底层配置类，直接指定Parquet序列化器与Glue表信息：

S3Bucket s3Bucket = new S3Bucket(this, "FirehoseDestBucket");

// 构造格式转换核心配置
CfnDeliveryStream.DataFormatConversionConfigurationProperty formatConversion = CfnDeliveryStream.DataFormatConversionConfigurationProperty.builder()
        .enabled(true)
        .inputFormatConfiguration(CfnDeliveryStream.InputFormatConfigurationProperty.builder()
                .deserializer(CfnDeliveryStream.DeserializerProperty.builder()
                        .openXJsonSerDe(CfnDeliveryStream.OpenXJsonSerDeProperty.builder().build()) // 按实际输入格式调整，如CSV用CsvSerDe
                        .build())
                .build())
        .outputFormatConfiguration(CfnDeliveryStream.OutputFormatConfigurationProperty.builder()
                .serializer(CfnDeliveryStream.SerializerProperty.builder()
                        .parquetSerDe(CfnDeliveryStream.ParquetSerDeProperty.builder()
                                .compression("SNAPPY") // 可选压缩格式：SNAPPY/GZIP/UNCOMPRESSED
                                .build())
                        .build())
                .build())
        .schemaConfiguration(CfnDeliveryStream.SchemaConfigurationProperty.builder()
                .databaseName(database.getDatabaseName())
                .roleArn(glueAccessRole.getRoleArn())
                .tableName(table.getTableName())
                .region(Stack.of(this).getRegion())
                .build())
        .build();

// 构造S3目标完整配置
CfnDeliveryStream.S3DestinationConfigurationProperty s3DestConfig = CfnDeliveryStream.S3DestinationConfigurationProperty.builder()
        .bucketArn(s3Bucket.getBucketArn())
        .roleArn(firehoseRole.getRoleArn())
        .dataFormatConversionConfiguration(formatConversion)
        .prefix("year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/")
        .errorOutputPrefix("errors/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/")
        .build();

// 创建Firehose投递流
CfnDeliveryStream firehose = new CfnDeliveryStream(this, "ParquetDeliveryStream", CfnDeliveryStreamProps.builder()
        .deliveryStreamType("DirectPut")
        .s3DestinationConfiguration(s3DestConfig)
        .build());

3. 配置IAM角色权限 #

确保Firehose角色拥有访问S3、Glue数据库与表的权限：

Role firehoseRole = new Role(this, "FirehoseExecutionRole", RoleProps.builder()
        .assumedBy(new ServicePrincipal("firehose.amazonaws.com"))
        .build());

// S3访问权限
firehoseRole.addToPolicy(new PolicyStatement(PolicyStatementProps.builder()
        .actions(List.of("s3:PutObject", "s3:GetBucketLocation"))
        .resources(List.of(s3Bucket.getBucketArn(), s3Bucket.getBucketArn() + "/*"))
        .build()));

// Glue访问权限
firehoseRole.addToPolicy(new PolicyStatement(PolicyStatementProps.builder()
        .actions(List.of("glue:GetTable", "glue:GetDatabase", "glue:GetPartition"))
        .resources(List.of(database.getDatabaseArn(), table.getTableArn()))
        .build()));

关键注意事项 #

• 输入格式需与Deserializer配置匹配，若输入为CSV，需替换为CsvSerDe。
• Glue表结构必须与Firehose转换后的数据结构完全一致，否则会导致转换失败。
• 此手动配置方式兼容后续CDK正式版本，待S3Bucket.processor实现后可平滑切换到高层封装。

内容的提问来源于stack exchange，提问作者Nicklas Karlsson