安装部署

当前inlong-sort是基于flink的一个应用,因此运行inlong-sort应用前,需要准备好flink 环境

由于当前inlong-sort依赖的是flink1.13.5版本,因此在下载部署包时,请选择flink-1.13.5-bin-scala_2.11.tgz

flink环境配置完成后,可以通过浏览器访问flink的web ui,对应的地址是/{flink部署路径}/conf/masters文件中的地址

准备安装文件

安装文件在inlong-sort目录。

启动inlong-sort应用

有了上述编译阶段产出的jar包后,就可以启动inlong-sort的应用了。提交方式可以参考如何提交flink作业

示例:

  1. ./bin/flink run -c org.apache.inlong.sort.singletenant.flink.Entrance inlong-sort/sort-[version].jar \
  2. --cluster-id debezium2hive --dataflow.info.file /YOUR_DATAFLOW_INFO_DIR/debezium-to-hive.json \
  3. --source.type pulsar --sink.type hive --sink.hive.rolling-policy.rollover-interval 60000 \
  4. --metrics.audit.proxy.hosts 127.0.0.1:10081 --sink.hive.rolling-policy.check-interval 30000

注意:

  • -c org.apache.inlong.sort.singletenant.flink.Entrance 表示main class name

  • inlong-sort/sort-[version].jar 为编译阶段产出的jar包

必要的配置

  • --cluster-id 用来唯一标识一个inlong-sort作业,同inlong-manager中sort.appName配置一致
  • --dataflow.info.file 流配置文件路径
  • --source.type 数据源的种类, 当前支持:”pulsar”
  • --sink.type 存储系统的种类,当前支持:”clickhouse”、”hive”、”iceberg”、”kafka”
  • --metrics.audit.proxy.hosts audit proxy 地址用于上报审计指标数据

启动参数配置示例

  1. --cluster-id debezium2kafka-canal --dataflow.info.file /YOUR_DATAFLOW_INFO_DIR/debezium-to-kafka-canal.json \
  2. --source.type pulsar --sink.type kafka

所有支持的配置

配置名是否必须默认值描述
cluster-idYNA用来唯一标识一个inlong-sort作业
source.typeYNA数据源的种类, 当前支持”pulsar”
sink.typeYNA存储系统的种类,当前支持”clickhouse”、”hive”、”iceberg”和”kafka”
source.parallelismN1source的并行度
deserialization.parallelismN1deserialization的并行度
transformation.parallelismN1transformation的并行度
sink.parallelismN1sink的并行度
checkpoint.intervalN600000checkpoint间隔,单位:毫秒
min.pause.between.checkpoints.msN500checkpoint之间的最小间隔,单位:毫秒
checkpoint.timeout.msN600000checkpoint超时时间,单位:毫秒
sink.field.type.string.nullableNfalsestring类型的sink field是否可以为空
sink.field.type.int.nullableNtrueint类型的sink field是否可以为空
sink.field.type.short.nullableNtrueshort类型的sink field是否可以为空
sink.field.type.long.nullableNtruelong类型的sink field是否可以为空
sink.hive.rolling-policy.file-sizeN134217728写hive时的文件滚动大小,单位:字节
sink.hive.rolling-policy.rollover-intervalN1800000写hive时的文件滚动时间间隔,单位:毫秒
sink.hive.rolling-policy.check-intervalN60000写hive时的文件滚动检查间隔,单位:毫秒