如何在AWS CDK的Kinesis Firehose配置中编写动态分区JQ表达式

问题背景 #

你有一个用于Kinesis Firehose动态分区的JQ表达式，作用是将格式为"time": "2020-01-29T17:26:50Z"的时间字段转换为20200129格式的字符串，作为S3存储的分区日期。原始JQ表达式如下：

time: .time | strptime("%Y-%m-%dT%H:%M:%SZ") | mktime | strftime("%Y%m%d")

但在编写AWS CDK（Python）代码时，配置MetadataExtractionQuery参数时遇到引号转义的问题，你的代码片段如下：

s3_destination_conf = ds.ExtendedS3DestinationConfigurationProperty(
    bucket_arn=output_bucket.bucket_arn,
    dynamic_partitioning_configuration=ds.DynamicPartitioningConfigurationProperty(
        enabled=True
    ),
    processing_configuration=ds.ProcessingConfigurationProperty(
        enabled=True,
        processors=[
            ds.ProcessorProperty(
                type="MetadataExtraction",
                parameters=[
                    ds.ProcessorParameterProperty(
                        parameter_name="JsonParsingEngine",
                        parameter_value="JQ-1.6",
                    ),
                    ds.ProcessorParameterProperty(
                        parameter_name="MetadataExtractionQuery",
                        # 问题行：引号转义错误导致JQ表达式无法正确执行
                        parameter_value="{time: .time| strptime('%Y-%m-%dT%H:%M:%SZ') | mktime | strftime('%Y%m%d')}", 
                    ),
                ],
            )
        ],
    ),
    role_arn=role.role_arn,
    buffering_hints=ds.BufferingHintsProperty(
        interval_in_seconds=params.KinesisFirehose.BUFFER_INTERVAL_SEC,
        size_in_m_bs=params.KinesisFirehose.BUFFER_SIZE_MB,
    ),
)

解决方案 #

核心问题是Python字符串的引号转义规则与JQ语法要求的冲突：JQ的strptime和strftime函数需要用双引号包裹格式字符串，而Python中如果用双引号包裹整个表达式，内部的双引号必须转义；或者用单引号包裹整个表达式，内部双引号无需转义。

以下是几种正确的写法：

写法1：用单引号包裹整个JQ表达式（推荐，无需转义） #

parameter_value='{time: .time | strptime("%Y-%m-%dT%H:%M:%SZ") | mktime | strftime("%Y%m%d")}'

写法2：用双引号包裹并转义内部双引号 #

parameter_value="{time: .time | strptime(\"%Y-%m-%dT%H:%M:%SZ\") | mktime | strftime(\"%Y%m%d\")}"

写法3：用三重引号包裹（更易读，适合复杂表达式） #

parameter_value="""{time: .time | strptime("%Y-%m-%dT%H:%M:%SZ") | mktime | strftime("%Y%m%d")}"""

修正后的完整代码片段 #

s3_destination_conf = ds.ExtendedS3DestinationConfigurationProperty(
    bucket_arn=output_bucket.bucket_arn,
    dynamic_partitioning_configuration=ds.DynamicPartitioningConfigurationProperty(
        enabled=True
    ),
    processing_configuration=ds.ProcessingConfigurationProperty(
        enabled=True,
        processors=[
            ds.ProcessorProperty(
                type="MetadataExtraction",
                parameters=[
                    ds.ProcessorParameterProperty(
                        parameter_name="JsonParsingEngine",
                        parameter_value="JQ-1.6",
                    ),
                    ds.ProcessorParameterProperty(
                        parameter_name="MetadataExtractionQuery",
                        # 使用推荐的单引号包裹写法
                        parameter_value='{time: .time | strptime("%Y-%m-%dT%H:%M:%SZ") | mktime | strftime("%Y%m%d")}', 
                    ),
                ],
            )
        ],
    ),
    role_arn=role.role_arn,
    buffering_hints=ds.BufferingHintsProperty(
        interval_in_seconds=params.KinesisFirehose.BUFFER_INTERVAL_SEC,
        size_in_m_bs=params.KinesisFirehose.BUFFER_SIZE_MB,
    ),
)

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