CREATE TABLE t1(
uuid VARCHAR(20), -- you can use 'PRIMARY KEY NOT ENFORCED' syntax to mark the field as record key
name VARCHAR(10),
age INT,
ts TIMESTAMP(3),
partition
VARCHAR(20)
)
PARTITIONED BY (partition
)
WITH (
'connector' = 'hudi',
'path' = '/data/t1',
'write.tasks' = '1', -- default is 4 ,required more resource
'compaction.tasks' = '1', -- default is 10 ,required more resource
'table.type' = 'COPY_ON_WRITE', -- this creates a MERGE_ON_READ table, by default is COPY_ON_WRITE
'read.tasks' = '1', -- default is 4 ,required more resource
'read.streaming.enabled' = 'true', -- this option enable the streaming read
'read.streaming.start-commit' = '0', -- specifies the start commit instant time
'read.streaming.check-interval' = '4' -- specifies the check interval for finding new source commits, default 60s.
);
-- insert data using values INSERT INTO t1 VALUES ('id1','Danny',23,TIMESTAMP '1970-01-01 00:00:01','par1'), ('id2','Stephen',33,TIMESTAMP '1970-01-01 00:00:02','par1'), ('id3','Julian',53,TIMESTAMP '1970-01-01 00:00:03','par2'), ('id4','Fabian',31,TIMESTAMP '1970-01-01 00:00:04','par2'), ('id5','Sophia',18,TIMESTAMP '1970-01-01 00:00:05','par3'), ('id6','Emma',20,TIMESTAMP '1970-01-01 00:00:06','par3'), ('id7','Bob',44,TIMESTAMP '1970-01-01 00:00:07','par4'), ('id8','Han',56,TIMESTAMP '1970-01-01 00:00:08','par4');
SELECT * FROM t1;
CREATE CATALOG datasource WITH ( 'type'='jdbc', 'property-version'='1', 'base-url'='jdbc:postgresql://postgres:5432/', 'default-database'='postgres', 'username'='postgres', 'password'='postgres' );
CREATE DATABASE IF NOT EXISTS datasource;
CREATE TABLE datasource.accident_claims WITH (
'connector' = 'kafka',
'topic' = 'pg_claims.claims.accident_claims',
'properties.bootstrap.servers' = 'kafka:9092',
'properties.group.id' = 'accident_claims-consumer-group',
'format' = 'debezium-json',
'scan.startup.mode' = 'earliest-offset'
) LIKE datasource.postgres.claims.accident_claims
(EXCLUDING ALL);
SELECT * FROM datasource.accident_claims;
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-postgres.json curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://localhost:8083/connectors/ -d @register-postgres-members.json
curl -i -X PUT -H "Accept: application/json" -H "Content-Type: application/json"
http://localhost:8083/connectors/claims-connector/config
-d @register-postgres.json
curl -i -X DELETE http://localhost:8083/connectors/members-connector curl -i -X DELETE http://localhost:8083/connectors/claims-connector
curl -X POST http://localhost:8083/connectors/members-connector/restart curl -X POST http://localhost:8083/connectors/claims-connector/restart
GET http://localhost:8083/connectors
SELECT * FROM pg_replication_slots;
SELECT pg_drop_replication_slot('slot_member');
docker compose exec kafka /kafka/bin/kafka-console-consumer.sh
--bootstrap-server kafka:9092
--from-beginning
--property print.key=true
--topic pg_claims.claims.members
docker compose exec kafka /kafka/bin/kafka-console-consumer.sh
--bootstrap-server kafka:9092
--from-beginning
--property print.key=true
--topic pg_claims.claims.members
INSERT INTO dwd.accident_claims SELECT claim_id, claim_total, claim_total_receipt, claim_currency, member_id, CAST(accident_date AS DATE), accident_type, accident_detail, CAST(claim_date AS DATE), -- Cast claim_date to DATE claim_status, CAST(ts_created AS TIMESTAMP), CAST(ts_updated AS TIMESTAMP), CAST(claim_date AS DATE) -- Cast claim_date to DATE FROM datasource.accident_claims;
INSERT INTO dwb.accident_claims SELECT claim_id, claim_total, claim_total_receipt, claim_currency, member_id, accident_date, accident_type, accident_detail, claim_date, claim_status, ts_created, ts_updated, claim_date FROM dwd.accident_claims;
{"claim_id": 100, "claim_total": null, "claim_total_receipt": null, "claim_currency": null, "member_id": null, "accident_date": null, "accident_type": null, "accident_detail": null, "claim_date": null, "claim_status": null, "ts_created": null, "ts_updated": null, "ds": null}
INSERT INTO dwd.members SELECT id, first_name, last_name, address, address_city, address_country, insurance_company, insurance_number, CAST(ts_created AS TIMESTAMP(3)), -- Cast ts_created to TIMESTAMP(3) CAST(ts_updated AS TIMESTAMP(3)), -- Ensure ts_updated is also TIMESTAMP(3) CAST(ts_created AS DATE) -- Cast ts_created to DATE FROM datasource.members;
docker compose exec postgres env PGOPTIONS="--search_path=claims" bash -c 'psql -U $POSTGRES_USER postgres'
PGOPTIONS="--search_path=claims"
psql -U $POSTGRES_USER -d postgres -a -f postgres_bootstrap.sql
docker compose exec 2563476e858f env PGOPTIONS="--search_path=claims" bash -c 'psql -U $POSTGRES_USER -d postgres -a -f /postgres/postgres_bootstrap.sql'
SELECT slot_name FROM pg_replication_slots;
curl -X POST http://CONNECT_REST_URL/connectors/CONNECTOR_NAME/restart
./bin/flink cancel -a