使用Flink将Hudi表写入S3失败，咨询错误原因与可行性

尝试通过Flink-CDC捕获MySQL数据变更，并将数据更新至S3存储的Hudi表，PyFlink作业代码如下：

env = StreamExecutionEnvironment.get_execution_environment(config)
env.set_parallelism(1)
settings = EnvironmentSettings.new_instance().in_streaming_mode().build()
t_env = StreamTableEnvironment.create(env, environment_settings=settings)

t_env.execute_sql(f"""
            CREATE TABLE source (
                ...
            ) WITH (
                'connector' = 'mysql-cdc',
                ...
            )
        """)
t_env.execute_sql(f"""
            CREATE TABLE target (
                ...
            ) WITH (
                'connector' = 'hudi',
                'path' = 's3a://xxx/xx/xx',
                'table.type' = 'COPY_ON_WRITE',
                'write.precombine.field' = 'id',
                'write.operation' = 'upsert',
                'hoodie.datasource.write.recordkey.field' = 'id',
                'hoodie.datasource.write.partitionpath.field' = '',
                'write.tasks' = '1',
                'compaction.tasks' = '1',
                'compaction.async.enabled' = 'true',
                'hoodie.table.version' = 'SIX',
                'hoodie.write.table.version' = '6'
            )
        """)
t_env.execute_sql(f"""
            INSERT INTO target
            SELECT * FROM source
        """)

提交作业后返回错误：

org.apache.hudi.exception.HoodieLockException: Unsupported scheme :s3a, since this fs can not support atomic creation

错误含义 #

这个错误的核心是Hudi默认依赖POSIX风格的原子文件创建能力实现分布式锁机制，但S3（通过s3a协议访问）是对象存储系统，不支持POSIX文件系统的原子操作（比如原子创建锁文件、原子重命名），导致Hudi的锁初始化失败。

S3是否支持作为Hudi Sink？Flink能否向S3执行Upsert？ #

S3完全可以作为Hudi的存储介质，Flink也能向S3执行Upsert操作，只是需要调整Hudi的锁配置，替换为适合S3环境的锁实现，而非默认的文件系统锁。

解决方案 #

需要在Hudi目标表的WITH参数中添加锁相关配置，以下是几种可行方案：

方案一：使用ZooKeeper作为锁管理器（通用分布式场景） #

添加如下参数：

'hoodie.write.lock.provider' = 'org.apache.hudi.client.transaction.lock.ZookeeperBasedLockProvider',
'hoodie.write.lock.zookeeper.url' = '你的ZK集群地址:端口',
'hoodie.write.lock.zookeeper.lock_key' = 'hudi-s3-lock',
'hoodie.write.lock.zookeeper.base_path' = '/hudi/locks'

方案二：使用DynamoDB作为锁管理器（AWS云环境） #

适合AWS部署的场景，添加如下参数：

'hoodie.write.lock.provider' = 'org.apache.hudi.client.transaction.lock.DynamoDBBasedLockProvider',
'hoodie.write.lock.dynamodb.table' = '你的DynamoDB锁表名称',
'hoodie.write.lock.dynamodb.region' = '你的AWS区域'

方案三：使用乐观锁（Hudi 0.12+版本，低并发场景） #

无需外部依赖，适合写入并发较低的场景，添加：

'hoodie.write.lock.provider' = 'org.apache.hudi.client.transaction.lock.OptimisticLockProvider'

注意：高并发写入时可能出现冲突，需要业务侧处理重试逻辑。

额外注意事项 #

• 确保Flink集群已包含对应锁实现的依赖包（比如ZK锁需要Hudi客户端及ZK相关依赖）；
• COPY_ON_WRITE表的compaction操作同样依赖锁机制，锁配置必须生效；
• 也可以尝试将S3协议从s3a://切换为s3://（基于AWS SDK），但锁配置的调整仍是必要步骤。

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