Sparksql and flinksql create and link table records

start-up flink sql:bin/sql-client.sh

Build table

flink Create a table

create table iceberg.xxx.xxx
(
    id       STRING comment 'id',
    dt       STRING comment ' Partition field '
)
    PARTITIONED BY (dt)
with (
    'write.format.default' = 'parquet', -- Specify the file storage format , Default parquet
    'write.parquet.compression-codec' = 'gzip', -- Specify the file compression format ,
    'commit.manifest-merge.enabled' = 'true', -- Auto merge on write manifest
    'history.expire.max-snapshot-age-ms' = '43200000', -- Historical snapshot retention time （ms）, Default 5 God , here 12h
    'engine.hive.enabled' = 'true', -- Support hive Inquire about 
    'write.metadata.delete-after-commit.enabled' = 'true', -- Delete the oldest metadata file after each new metadata file is created 
    'write.metadata.previous-versions-max' = '20', -- The maximum number of metadata files of previous versions deleted after submission 
    'write.metadata.compression-codec' = 'gzip', -- Enable metadata compression as gzip Format 
    'location' = 'hdfs://ns1/lakehouse/schema_name/table_name' -- Appoint hdfs Address 
);

spark Create a table

 create table iceberg.xxx.xxxx
(
    id     STRING comment 'id',
    dt       STRING comment ' Partition field '
)
using iceberg
partitioned by (dt)
location 'hdfs://xxx/lakehouse/schema_name/table_name'
tblproperties (
'write.format.default' = 'parquet',                    -- Specify the file storage format , Default parquet
'write.parquet.compression-codec' = 'gzip',            -- Specify the file compression format ,
'commit.manifest-merge.enabled' = 'true',              -- Auto merge on write manifest
'history.expire.max-snapshot-age-ms' = '43200000',     -- Historical snapshot retention time （ms）, Default 5 God , here 12h
'engine.hive.enabled' = 'true',                        -- Support hive Inquire about 
'write.metadata.delete-after-commit.enabled' = 'true', -- Delete the oldest metadata file after each new metadata file is created 
'write.metadata.previous-versions-max' = '20',         -- The maximum number of metadata files of previous versions deleted after submission 
'write.metadata.compression-codec' = 'gzip'            -- Enable metadata compression as gzip Format 
);

Create a target table based on the source table

CREATE TABLE iceberg.schema_name.table_name
    using iceberg
    partitioned by (dt)
    location 'hdfs://ns1/lakehouse/schema_name/table_name'
    tblproperties (
    'write.format.default' = 'parquet',
    'write.parquet.compression-codec' = 'gzip',
    'commit.manifest-merge.enabled' = 'true',
    'engine.hive.enabled' = 'true',
    'write.metadata.delete-after-commit.enabled' = 'true',
    'write.metadata.previous-versions-max' = '20',
    'write.metadata.compression-codec' = 'gzip'
    ) AS SELECT * from iceberg.schema_name.original_table_name where 1=2;

Connect mysql、 sql server

spark

-- mysql
CREATE
TEMPORARY
VIEW tb_order_group
USING org.apache.spark.sql.jdbc
OPTIONS (
  url 'jdbc:mysql://xx.xxx.xx.xx:4909/db_name?serverTimezone=GMT%2B8&useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&zeroDateTimeBehavior=convertToNull',
  dbtable 'xxx',
  user 'xxx',
  password 'xxx'
);

-- sqlserver
CREATE
TEMPORARY
VIEW gxywhz
USING org.apache.spark.sql.jdbc
OPTIONS (
  url 'jdbc:sqlserver://192.168.1.xx:xxx;DatabaseName=xxx',
  dbtable 'dbo.xxx',
  user 'xxx',
  password 'xxxx'
);

flink

-- Connector Connect   Create a mapping table 
 CREATE TABLE mysql_source
(
    id          int comment 'id'
    primary key (id) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://192.168.xx.xx:3306/xxx',
    'table-name' = 'tableName',
    'driver' = 'com.mysql.jdbc.Driver',
    'username' = 'root',
    'password' = 'xxx'
);

CREATE TABLE sqlserver_source
(
    id     STRING comment 'id'
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:jtds:sqlserver://192.168.xx.xxx:10009;databaseName=xxx;useLOBs=false',
'table-name' = 'schema.tableName',
'driver' = 'net.sourceforge.jtds.jdbc.Driver',
'username' = 'xx',
'password' = 'xxxxx'
 );