sparksql 与flinksql 建表 与 连表记录

启动flink sql:bin/sql-client.sh

建表

flink建立表

create table iceberg.xxx.xxx
(
    id       STRING comment 'id',
    dt       STRING comment '分区字段'
)
    PARTITIONED BY (dt)
with (
    'write.format.default' = 'parquet', --指定文件存储格式，默认parquet
    'write.parquet.compression-codec' = 'gzip', --指定文件压缩格式，
    'commit.manifest-merge.enabled' = 'true', --写入时自动合并manifest
    'history.expire.max-snapshot-age-ms' = '43200000', --历史快照保留时间（ms），默认5天,此处12h
    'engine.hive.enabled' = 'true', --支持hive查询
    'write.metadata.delete-after-commit.enabled' = 'true', --在每个新元数据文件创建后删除最旧的元数据文件
    'write.metadata.previous-versions-max' = '20', --提交后删除之前版本元数据文件的最大数量
    'write.metadata.compression-codec' = 'gzip', --开启元数据压缩为gzip格式
    'location' = 'hdfs://ns1/lakehouse/schema_name/table_name' --指定hdfs地址
);

spark建立表

 create table iceberg.xxx.xxxx
(
    id     STRING comment 'id',
    dt       STRING comment '分区字段'
)
using iceberg
partitioned by (dt)
location 'hdfs://xxx/lakehouse/schema_name/table_name'
tblproperties (
'write.format.default' = 'parquet',                    --指定文件存储格式，默认parquet
'write.parquet.compression-codec' = 'gzip',            --指定文件压缩格式，
'commit.manifest-merge.enabled' = 'true',              --写入时自动合并manifest
'history.expire.max-snapshot-age-ms' = '43200000',     --历史快照保留时间（ms），默认5天,此处12h
'engine.hive.enabled' = 'true',                        --支持hive查询
'write.metadata.delete-after-commit.enabled' = 'true', --在每个新元数据文件创建后删除最旧的元数据文件
'write.metadata.previous-versions-max' = '20',         --提交后删除之前版本元数据文件的最大数量
'write.metadata.compression-codec' = 'gzip'            --开启元数据压缩为gzip格式
);

根据源表创建目标表

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;

连接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连接 创建映射表
 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'
 );