小编Har*_*h J的帖子

我在 Terraform 中有一个 Kinesis Firehose 配置，它从 JSON 格式的 Kinesis 流读取数据，使用 Glue 将其转换为 Parquet 并写入 S3。数据格式转换出现问题，我收到以下错误（删除了一些详细信息）：

{“attemptsMade”：1，“arrivalTimestamp”：1624541721545，“lastErrorCode”：“DataFormatConversion.InvalidSchema”，“lastErrorMessage”：“架构无效。指定的表没有列。”，“attemptEndingTimestamp”：1624542026951，“rawData ":"xx","sequenceNumber":"xx","subSequenceNumber":null,"dataCatalogTable":{"catalogId":null,"databaseName":"db_name","tableName":"table_name","re​​gion" :null,"versionId":"最新","roleArn":"xx"}}

我正在使用的 Glue Table 的 Terraform 配置如下：

resource "aws_glue_catalog_table" "stream_format_conversion_table" {
  name          = "${var.resource_prefix}-parquet-conversion-table"
  database_name = aws_glue_catalog_database.stream_format_conversion_db.name

  table_type = "EXTERNAL_TABLE"

  parameters = {
    EXTERNAL              = "TRUE"
    "parquet.compression" = "SNAPPY"
  }

  storage_descriptor {
    location      = "s3://${element(split(":", var.bucket_arn), 5)}/"
    input_format  = "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat"
    output_format = "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat"

    ser_de_info {
      name                  = "my-stream"
      serialization_library = "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe"

      parameters = {
        "serialization.format" = 1
      }
    }
    columns {
      name = …

我在 AWS Glue 环境中工作。我从 Glue 目录中读取数据作为动态数据帧，并将其转换为 Pyspark 数据帧以进行自定义转换。为了更新插入新的/更新的数据，我打算使用增量表。

但我只找到从路径读取数据作为增量表的选项。我需要将 Pyspark 数据帧转换为 Delta 表以进行合并操作。有什么办法可以做到这一点吗？