从智能工厂到车联网:S3 Tables 双模式写入实战指南

import jsonimport pandas as pdimport pyarrow as pafrom pyiceberg.catalog import load_catalog
def lambda_handler(event, context):
    """
    Lambda 函数：批量写入 IoT 数据到 S3 Tables
    
    Event 格式：
    {
        "records": [
            {"id": 1, "sensor_name": "temp_01", "value": 25.5, "timestamp": "2024-01-01T10:00:00Z"},
            {"id": 2, "sensor_name": "temp_02", "value": 26.1, "timestamp": "2024-01-01T10:01:00Z"}
        ]
    }
    """
try:
        # 1. 解析输入数据
        records = event.get('records', [])
        if not records:
            return {
                'statusCode': 400,
                'body': json.dumps({'error': 'No records provided'})
            }
        
        # 2. 连接 S3 Tables Catalog
        catalog = load_catalog("s3tables", **{
            "type": "rest",
            "warehouse": "arn:aws:s3tables:us-east-1:ACCOUNT_ID:bucket/BUCKET_NAME",
            "uri": "https://s3tables.us-east-1.amazonaws.com/iceberg",
            "rest.sigv4-enabled": "true",
            "rest.signing-name": "s3tables",
            "rest.signing-region": "us-east-1",
        })
        
        # 3. 加载表
        table = catalog.load_table("iot_namespace.sensor_data")
        
        # 4. 数据转换
        df = pd.DataFrame(records)
        
        # 确保数据类型正确
        df['id'] = df['id'].astype('int32')
        df['value'] = df['value'].astype('float64')
        df['timestamp'] = pd.to_datetime(df['timestamp'])
        
        # 转换为 PyArrow Table
        arrow_table = pa.Table.from_pandas(df)
        
        # 转换时间戳类型为微秒精度UTC
        if 'timestamp' in arrow_table.column_names:
            timestamp_col = arrow_table.column('timestamp')
            new_timestamp = pa.compute.cast(timestamp_col, pa.timestamp('us', tz='UTC'))
            arrow_table = arrow_table.set_column(
                arrow_table.schema.get_field_index('timestamp'),
                'timestamp',
                new_timestamp
            )
        
        # 5. 写入 S3 Tables
        table.append(arrow_table)
        
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': f'Successfully wrote {len(records)} records',
                'records_count': len(records)
            })
        }
        
    except Exception as e:
        print(f"Error: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

pyiceberg[s3fs,glue]==0.7.0
pandas==2.1.0
pyarrow==14.0.0

{
    "Version": "2012-10-17",
    "Statement": [
      {
        "Effect": "Allow",
        "Action": [
          "s3tables:GetTable",
          "s3tables:UpdateTableMetadata",
          "s3tables:PutTableData"
        ],
        "Resource": "arn:aws:s3tables:*:*:bucket/*/table/*"
      },
      {
        "Effect": "Allow",
        "Action": [
          "s3:PutObject",
          "s3:GetObject"
        ],
        "Resource": "arn:aws:s3:::your-bucket/*"
      }
    ]
  }

POST /iot/batch-upload

curl -X POST https://api-id.execute-api.us-east-1.amazonaws.com/prod/iot/batch-upload \
  -H "Content-Type: application/json" \
  -H "x-api-key: YOUR_API_KEY" \
  -d '{
    "records": [
      {
        "id": 1,
        "sensor_name": "temperature_01",
        "value": 25.5,
        "timestamp": "2024-01-01T10:00:00Z"
      },
      {
        "id": 2,
        "sensor_name": "humidity_01",
        "value": 60.2,
        "timestamp": "2024-01-01T10:01:00Z"
      }
    ]
  }'

**使用 AWS CLI 部署**：

# 1. 创建 IAM 角色
aws iam create-role \
    --role-name s3tables-lambda-role \
    --assume-role-policy-document '{
      "Version": "2012-10-17",
      "Statement": [{
        "Effect": "Allow",
        "Principal": {"Service": "lambda.amazonaws.com"},
        "Action": "sts:AssumeRole"
      }]
    }'

# 2. 附加权限策略
aws iam attach-role-policy \
    --role-name s3tables-lambda-role \
    --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

aws iam put-role-policy \
    --role-name s3tables-lambda-role \
    --policy-name s3tables-access \
    --policy-document '{
      "Version": "2012-10-17",
      "Statement": [
        {
          "Effect": "Allow",
          "Action": ["s3tables:*"],
          "Resource": "*"
        },
        {
          "Effect": "Allow",
          "Action": ["s3:PutObject", "s3:GetObject"],
          "Resource": "arn:aws:s3:::your-bucket/*"
        }
      ]
    }'

# 3. 打包 Lambda 函数
cd lambda
pip install -t package pyiceberg[s3fs] pandas pyarrow
cd package && zip -r ../function.zip . && cd ..
zip -g function.zip lambda_function.py

# 4. 创建 Lambda 函数
aws lambda create-function \
    --function-name s3tables-iot-writer \
    --runtime python3.12 \
    --handler lambda_function.lambda_handler \
    --role arn:aws:iam::ACCOUNT_ID:role/s3tables-lambda-role \
    --zip-file fileb://function.zip \
    --timeout 60 \
    --memory-size 1024

# 5. 创建 API Gateway
API_ID=$(aws apigatewayv2 create-api \
    --name iot-s3tables-api \
    --protocol-type HTTP \
    --target arn:aws:lambda:us-east-1:ACCOUNT_ID:function:s3tables-iot-writer \
    --query 'ApiId' --output text)

# 6. 授权 API Gateway 调用 Lambda
aws lambda add-permission \
    --function-name s3tables-iot-writer \
    --statement-id apigateway-invoke \
    --action lambda:InvokeFunction \
    --principal apigateway.amazonaws.com \
    --source-arn "arn:aws:execute-api:us-east-1:ACCOUNT_ID:${API_ID}/*/*"

echo "API Endpoint: https://${API_ID}.execute-api.us-east-1.amazonaws.com/iot/batch-upload"

{
     "Version": "2012-10-17",
     "Statement": [
         {
             "Sid": "S3TableAccessViaGlueFederation",
             "Effect": "Allow",
             "Action": [
                 "glue:GetTable",
                 "glue:GetDatabase",
                 "glue:UpdateTable"
             ],
             "Resource": [
                "arn:aws:glue:ap-southeast-1:123456789012:catalog/s3tablescatalog/*",

                "arn:aws:glue:ap-southeast-1:123456789012:catalog/s3tablescatalog",

                "arn:aws:glue:ap-southeast-1:123456789012:catalog",

                "arn:aws:glue:ap-southeast-1:123456789012:database/*",
                "arn:aws:glue:ap-southeast-1:123456789012:table/*/*"
             ]
         },
         {
             "Sid": "S3DeliveryErrorBucketPermission",
             "Effect": "Allow",
             "Action": [
                 "s3:AbortMultipartUpload",
                 "s3:GetBucketLocation",
                 "s3:GetObject",
                 "s3:ListBucket",
                 "s3:ListBucketMultipartUploads",
                 "s3:PutObject"
             ],
             "Resource": "*"

        },

        {

            "Sid": "RequiredWhenUsingKinesisDataStreamsAsSource",

            "Effect": "Allow",

            "Action": [

                "kinesis:DescribeStream",

                "kinesis:GetShardIterator",

                "kinesis:GetRecords",

                "kinesis:ListShards"

            ],

            "Resource": "*"
         },
         {
             "Sid": "RequiredWhenDoingMetadataReadsANDDataAndMetadataWriteViaLakeformation",
             "Effect": "Allow",
             "Action": [
                 "lakeformation:GetDataAccess"
             ],
             "Resource": "*"
         },
         {
             "Sid": "RequiredWhenUsingKMSEncryptionForS3ErrorBucketDelivery",
             "Effect": "Allow",
             "Action": [
                 "kms:Decrypt",
                 "kms:GenerateDataKey"
             ],
             "Resource": [
                 "*"

            ],

            "Condition": {

                "StringEquals": {

                    "kms:ViaService": "s3.ap-southeast-1.amazonaws.com"

                },

                "StringLike": {

                    "kms:EncryptionContext:aws:s3:arn": "*"
                 }
             }
         },
         {
             "Sid": "LoggingInCloudWatch",
             "Effect": "Allow",
             "Action": [
                 "logs:PutLogEvents"
             ],
             "Resource": "*"

        },

        {

            "Sid": "RequiredWhenAttachingLambdaToFirehose",

            "Effect": "Allow",

            "Action": [

                "lambda:InvokeFunction",

                "lambda:GetFunctionConfiguration"

            ],

            "Resource": "*"
         }
     ]
 }

{
"id": {{random.number(99999)}},
"name": "{{random.arrayElement( ["CHECKIN","CHECKOUT","PAYMENT_FULL","PAYMENT_PARTIAL"] )}}",
"value": {{random.number(99999)}}
}