使用AWS Wrangler在Athena创建表时遇语法错误求助

问题描述 #

通过AWS Wrangler的wr.athena.read_sql_query执行CREATE TABLE IF NOT EXISTS的CTAS语句，创建Parquet格式（SNAPPY压缩）的分区统计表时，触发InvalidRequestException，提示输入不匹配，期望'('或查询语句。

原SQL语句 #

query = f"""CREATE TABLE IF NOT EXISTS cantidad_mediciones_por_cliente_sensor WITH (
        format = 'Parquet',
        write_compression = 'SNAPPY',
        external_location = 's3://{self.params['bucket']}/{self.params['path']}/',
        partitioned_by = ARRAY['year', 'month', 'day']
        ) AS
        SELECT count  (distinct (fecha_hora)) as "cantidad_mediciones",
            sensor.id_sensor as "id_sensor",
            sensor.descripcion_sensor as "descripcion_sensor",
            sensor.tag_sensor as "tag_sensor",
            tipo_dispositivo.tipo_sensor as "tipo_sensor",
            tipo_dispositivo.id as "id_tipo_sensor",
            cliente.nombre_cliente as "nombre_cliente",
            cliente.id_cliente as "id_cliente",
            partition_0 as "year",
            partition_1 as "month",
            partition_2 as "day"
        FROM "{self.BASE_DE_DATOS}"."{self.TABLA}"
        WHERE (CAST(partition_0 as smallint) = {self.yesterday.year})
            and (CAST(partition_1 as tinyint) = {self.yesterday.month})
            and (CAST (partition_2 as tinyint) = {self.yesterday.day})
        GROUP BY sensor.id_sensor,
            sensor.descripcion_sensor,
            sensor.tag_sensor,
            tipo_dispositivo.tipo_sensor,
            tipo_dispositivo.id,
            cliente.id_cliente,
            cliente.nombre_cliente,
            partition_0,    
            partition_1,
            partition_2"""

错误信息 #

{
  "errorMessage": "An error occurred (InvalidRequestException) when calling the StartQueryExecution operation: line 4:4: mismatched input 'CREATE'. Expecting: '(', <query>",
  "errorType": "InvalidRequestException",
  "stackTrace": [
    "  File \"/var/task/measurement_daily_src/lambda_function.py\", line 80, in lambda_handler\n    query_for_athena.obtener()\n",
    "  File \"/var/task/measurement_daily_src/lambda_function.py\", line 22, in obtener\n    df_resumen_diario = self._obtener_datos_desde_athena()\n",
    "  File \"/var/task/measurement_daily_src/lambda_function.py\", line 68, in _obtener_datos_desde_athena\n    return wr.athena.read_sql_query(query, self.BASE_DE_DATOS, workgroup=self.params['workgroup'])\n",
    "  File \"/opt/python/awswrangler/_config.py\", line 450, in wrapper\n    return function(**args)\n",
    "  File \"/opt/python/awswrangler/athena/_read.py\", line 854, in read_sql_query\n    return _resolve_query_without_cache(\n",
    "  File \"/opt/python/awswrangler/athena/_read.py\", line 422, in _resolve_query_without_cache\n    return _resolve_query_without_cache_ctas(\n",
    "  File \"/opt/python/awswrangler/athena/_read.py\", line 263, in _resolve_query_without_cache_ctas\n    ctas_query_info: Dict[str, Union[str, _QueryMetadata]] = create_ctas_table(\n",
    "  File \"/opt/python/awswrangler/_config.py\", line 450, in wrapper\n    return function(**args)\n",
    "  File \"/opt/python/awswrangler/athena/_utils.py\", line 821, in create_ctas_table\n    raise ex\n",
    "  File \"/opt/python/awswrangler/athena/_utils.py\", line 800, in create_ctas_table\n    query_execution_id: str = _start_query_execution(\n",
    "  File \"/opt/python/awswrangler/athena/_utils.py\", line 98, in _start_query_execution\n    response: Dict[str, Any] = _utils.try_it(\n",
    "  File \"/opt/python/awswrangler/_utils.py\", line 343, in try_it\n    return f(**kwargs)\n",
    "  File \"/var/runtime/botocore/client.py\", line 530, in _api_call\n    return self._make_api_call(operation_name, kwargs)\n",
    "  File \"/var/runtime/botocore/client.py\", line 960, in _make_api_call\n    raise error_class(parsed_response, operation_name)\n"
  ]
}

错误原因 #

wr.athena.read_sql_query的核心逻辑是自动将传入的SQL语句包装成一个临时CTAS查询，目的是把查询结果导出到S3再转成DataFrame返回。当你传入完整的CREATE TABLE ... AS SELECT语句时，Wrangler会生成嵌套的CTAS语法，类似：

CREATE TABLE temp_table AS (
    CREATE TABLE IF NOT EXISTS cantidad_mediciones_por_cliente_sensor WITH (...) AS SELECT ...
)

这种嵌套结构不符合Athena的SQL语法规则，因此触发语法错误。

解决方案 #

分两步操作：先执行CTAS语句创建目标表，再查询该表获取结果DataFrame。

修改后的代码示例 #

def _obtener_datos_desde_athena(self):
    # 1. 执行CTAS语句创建持久化分区表
    ctas_query = f"""CREATE TABLE IF NOT EXISTS cantidad_mediciones_por_cliente_sensor WITH (
            format = 'Parquet',
            write_compression = 'SNAPPY',
            external_location = 's3://{self.params['bucket']}/{self.params['path']}/',
            partitioned_by = ARRAY['year', 'month', 'day']
            ) AS
            SELECT count(DISTINCT fecha_hora) as "cantidad_mediciones",
                sensor.id_sensor as "id_sensor",
                sensor.descripcion_sensor as "descripcion_sensor",
                sensor.tag_sensor as "tag_sensor",
                tipo_dispositivo.tipo_sensor as "tipo_sensor",
                tipo_dispositivo.id as "id_tipo_sensor",
                cliente.nombre_cliente as "nombre_cliente",
                cliente.id_cliente as "id_cliente",
                partition_0 as "year",
                partition_1 as "month",
                partition_2 as "day"
            FROM "{self.BASE_DE_DATOS}"."{self.TABLA}"
            WHERE (CAST(partition_0 as smallint) = {self.yesterday.year})
                and (CAST(partition_1 as tinyint) = {self.yesterday.month})
                and (CAST(partition_2 as tinyint) = {self.yesterday.day})
            GROUP BY sensor.id_sensor,
                sensor.descripcion_sensor,
                sensor.tag_sensor,
                tipo_dispositivo.tipo_sensor,
                tipo_dispositivo.id,
                cliente.id_cliente,
                cliente.nombre_cliente,
                partition_0,    
                partition_1,
                partition_2"""
    
    # 执行CTAS语句，用execute_sql而非read_sql_query
    wr.athena.execute_sql(ctas_query, database=self.BASE_DE_DATOS, workgroup=self.params['workgroup'])
    
    # 2. 查询刚创建的表，获取结果DataFrame
    select_query = f"""SELECT * FROM cantidad_mediciones_por_cliente_sensor
                      WHERE year = {self.yesterday.year}
                        AND month = {self.yesterday.month}
                        AND day = {self.yesterday.day}"""
    
    return wr.athena.read_sql_query(select_query, self.BASE_DE_DATOS, workgroup=self.params['workgroup'])

额外优化点 #

• 原SQL中count (distinct (fecha_hora))的多余空格可以去掉，写成count(DISTINCT fecha_hora)，避免潜在语法问题。
• 执行CTAS后，如果表是分区表，建议调用wr.athena.repair_table刷新分区元数据：

  wr.athena.repair_table(
      table="cantidad_mediciones_por_cliente_sensor",
      database=self.BASE_DE_DATOS,
      workgroup=self.params['workgroup']
  )
  

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