MrDannyWu commented on issue #6021:
URL: https://github.com/apache/iceberg/issues/6021#issuecomment-1285297768
> Can you look at your sql?
Thank you for your reply and help, I will provide you with my steps.
The background of the problem is that I want to synchronize mysql data to
Iceberg (Hive Catalog) through Flink CDC. The default is to write to Iceberg in
Append mode. If there is update or delete, there will be duplicate data. If I
want to use Upsert mode, there is a problem. In fact, I just want to know how
to write Iceberg (Hive Catalog) through Upsert.
step 1: create table on hive
SET engine.hive.enabled=true;
SET iceberg.engine.hive.enabled=true;
SET iceberg.mr.catalog=hive;
SET hive.vectorized.execution.enabled=false;
add jar /soft/iceberg-hive-runtime-1.0.0.jar;
CREATE EXTERNAL TABLE ods_data_1_1(
`id` BIGINT,
`name` STRING,
`age` BIGINT,
`gender` STRING,
`amount` BIGINT,
`geohash_code` STRING,
`status` BIGINT,
`location` STRING,
PRIMARY KEY (`id`) NOT ENFORCED
)
STORED BY 'org.apache.iceberg.mr.hive.HiveIcebergStorageHandler'
LOCATION 'hdfs://nn-1:8020/user/hive/warehouse/danny_test/ods_data_1_1'
TBLPROPERTIES (
'format-version'='2',
'iceberg.mr.catalog'='hadoop',
'iceberg.mr.catalog.hadoop.warehouse.location'='hdfs://nn-1:8020/user/hive/warehouse/danny_test/ods_data_1_1',
'write.upsert.enabled'='true',
'write.metadata.delete-after-commit.enabled'='true',
'write.metadata.previous-versions-max'='2'
);
step 2:Flink CDC
SET execution.checkpointing.interval = 3s;
SET execution.result-mode=table;
SET execution.result-mode=tableau;
SET yarn.application.queue=root;
# create source table
CREATE TABLE ods_data_1_1 (
`id` BIGINT,
`name` STRING,
`age` BIGINT,
`gender` STRING,
`amount` BIGINT,
`geohash_code` STRING,
`status` BIGINT,
`location` STRING,
PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'x.x.x.x',
'port' = '3306',
'username' = 'name',
'password' = 'pass',
'database-name' = 'db1',
'table-name' = 'tb1'
);
# create catalog
CREATE CATALOG ice WITH (
'type'='iceberg',
'catalog-type'='hive',
'uri'='thrift://nn-1:9083',
'clients'='5',
'property-version'='1',
'warehouse'='hdfs://nn-1:8020/user/hive/warehouse/'
);
# submit job
insert into ice.danny_test.ods_data_1_1 select * from ods_data_1_1 ;
report error:
>[ERROR] Could not execute SQL statement. Reason:
java.lang.IllegalStateException: Equality field columns shouldn't be empty
when configuring to use UPSERT data stream.
# PS
Flink 14.4 iceberg 1.0.0 hive 3.x
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
