Kwritin

Flink 概念辨析 Operator, Task, Subtask, Slot

2025-07-29T00:00:00+00:00

记得刚开始学 Flink 的时候，看官方文档有时会被它的概念搞得头晕。今天我尝试在 Flink Dashboard 的帮助下快速厘清几个基本概念！

Operator, Task, Subtask, Slot。

首先可以先扫一遍官方文档中提到的这几个概念。

总结一下：

Flink 在部署的时候，会将 算子(Operator) 组成 算子链(Operator Chain) 也就是 Task ，每个 Task 运行在单独的线程中，这样做的好处是减少数据交换、缓冲区开销等。
Flink 提供的并行度机制，可以将 Task 转为多个并行的 SubTask 运行在各自的线程中。
每个 SubTask 都需要部署在 Slot 中运行。
来自同一个 Job 中的 SubTask 可以和除了自己的其它 Subtask 共享 Slot！

最后一点有点绕，就是说：一个 Job，虽然有多个 Subtask，但是它们可以部署在一个 Slot 中。所以就变成：一个 Slot 可以部署 Job 的一个并行度。

如图所示: 左侧 TM 的第一个 Slot 上部署了整个 Job。

如果说你看到这里还是有点懵，接下来我们自己创建一个 Demo 来理解。

创建 Demo

接下来我们准备一个 Demo 程序，并且提交到 Flink Standard Cluster 中:

代码如下:

KafkaSource<byte[]> kafkaSource = KafkaSource.<byte[]>builder()
  .setBootstrapServers(kafkaConfig.bootstrapServers)
  .setTopics(kafkaConfig.topicNames.heartbeat)
  ...
  .build();

DataStream<Heartbeat> dataStream = streamEnv
  .fromSource(
    kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource"
  )
  .map(new HeartbeatMapper()).name("heartbeatMap").uid("heartbeatMap")
  .filter(hb -> hb.getVin() != null).name("heartbeatFilter").uid("heartbeatFilter");

dataStream
  .shuffle() // 手动触发
  .addSink(new PrintSinkFunction<>()).uid("printSink").name("printSink");

代码非常简单: 从 Kafka 读取数据，然后进行 map, filter, shuffle 后通过 printSink 打印。

分析 Demo

在 Flink DataStream Api 中，Operator 就是算子，常见的有 Fliter/Map/Join 等。

上述代码的 HeartbeatMapper, Lambda Filter 都是 Operator。

Operator 是 User Api 层面(逻辑层) 的概念，最终运行的时候，会经过 Flink 框架一系列的优化，将 Operator 组成链:

可以看到，我们的作业被组成了 2 个链。组成算子链可以减少数据交换以带来性能提升。

算子链就是 Task，我们的作业包含两个 Task(蓝色方块)。每个 Task 都会交给 Slot 上运行。

此时任务并行度为 1，Flink 集群 Slot 数量为 1 即可部署。

当并行度改为 2 时，1 个 Task 变为 2 个 SubTask，我们修改 Flink 集群的配置:

taskmanager.numberOfTaskSlots: 2

然后也可以正常部署:

最后我们进入到 TaskManager 页面，进行 Thread Dump:

"Source Data Fetcher for Source: kafkaSource -> heartbeatMap -> heartbeatFilter (2/2)#0" Id=151 RUNNABLE (in native)
	at sun.nio.ch.KQueueArrayWrapper.kevent0(Native Method)
	at sun.nio.ch.KQueueArrayWrapper.poll(KQueueArrayWrapper.java:198)
	at sun.nio.ch.KQueueSelectorImpl.doSelect(KQueueSelectorImpl.java:117)
  ...

"Sink: printSink (2/2)#0" Id=142 TIMED_WAITING on java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject@6a776ec
	at sun.misc.Unsafe.park(Native Method)
	-  waiting on java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject@6a776ec
	at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2163)
	...

"Sink: printSink (1/2)#0" Id=141 TIMED_WAITING on java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject@648bd7b2
	at sun.misc.Unsafe.park(Native Method)
	-  waiting on java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject@648bd7b2
	at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	...

"Source: kafkaSource -> heartbeatMap -> heartbeatFilter (2/2)#0" Id=140 TIMED_WAITING on java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject@34bf2376
	at sun.misc.Unsafe.park(Native Method)
	-  waiting on java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject@34bf2376
	at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
	...

可以清楚地看到我们的 4 个 SubTask。

至此，Flink 算子运行时模型已经基本清楚了，那么现在问题来了，知道 Slot 有什么用呢？

在自研大数据平台的时候，往往是基于 K8S 对 Flink 作业进行调度，那么每个任务配置了多少并行度，理论上就需要多少个 Slot，也就是多少个 TM。

单 Slot TM 的隔离性最好，就是一个 TM 专属于某一个作业。

<<<EOF

Flink 内存配置实践

2025-07-09T00:00:00+00:00

记一下在部署 Flink 任务时遇到的内存配置问题。

本地测试环境搭建

OS: macOS Sequoia (arm64)
Java: 1.8 (Amazon Corretto)
Flink: 1.15.4
Local K8S: Kind + Podman
Arthas: 4.0.5

TaskManager

由于部署成功之后，我们的作业实际上运行在 tm 的 slot 里，所以我们优先看 tm 的内存配置。

我们先不管三七二十一，启动一个本地的 session 集群: bin/start-cluster.sh

然后通过 web ui 观察 tm 的内存模型:

大致介绍一下各个部分的内容:

JVM Heap:

Framework Heap: 框架堆内存, 用于 Flink 框架的 JVM 堆内存。
Task Heap: 任务堆内存, 用于 Flink 应用的算子及用户代码的 JVM 堆内存。

Off-Heap:

Managed Memory: 托管内存, 由 Flink 管理的用于排序、哈希表、缓存中间结果及 RocksDB State Backend 的本地内存。
Direct Memory:
- Framework Off-heap: 框架堆外内存, 用于 Flink 框架的堆外内存（直接内存或本地内存）。
- Task Off-heap: 任务堆外内存, 用于 Flink 应用的算子及用户代码的堆外内存（直接内存或本地内存）。
- Network: 网络内存, 用于任务之间数据传输的直接内存（例如网络传输缓冲）。
JVM Metaspace: Flink JVM 进程的 Metaspace。
JVM Overhead: JVM 开销, 用于其他 JVM 开销的本地内存，例如栈空间、垃圾回收空间等。

其中，标记背景颜色的部份需要根据任务数据量调整。

Flink 在内存配置上支持 3 种模式:

只配置进程内存: 适用于 Yarn/K8S 等资源调度模式，进程内存直接和 K8S 的 resource.limit 对应起来。
只配置 Flink 内存: 适合本地调试，起一个 standard session cluster。
配置所有组件的内存: 精细化地管理每个组件的内存，相对复杂。

对于前两者，Flink 虽然屏蔽了底层细节，但内部也是通过一系列的计算得到各个组件的内存配置:

org/apache/flink/runtime/util/config/memory/ProcessMemoryUtils.java

public CommonProcessMemorySpec<FM> memoryProcessSpecFromConfig(Configuration config) {
  if (options.getRequiredFineGrainedOptions().stream().allMatch(config::contains)) {
    // all internal memory options are configured, use these to derive total Flink and
    // process memory
    return deriveProcessSpecWithExplicitInternalMemory(config);
  } else if (config.contains(options.getTotalFlinkMemoryOption())) {
    // internal memory options are not configured, total Flink memory is configured,
    // derive from total flink memory
    return deriveProcessSpecWithTotalFlinkMemory(config);
  } else if (config.contains(options.getTotalProcessMemoryOption())) {
    // total Flink memory is not configured, total process memory is configured,
    // derive from total process memory
    return deriveProcessSpecWithTotalProcessMemory(config);
  }
  return failBecauseRequiredOptionsNotConfigured();
}

在只配置了进程内存(process mem)的情况下，flink 会先计算出 metaspace 和 overhead 的内存，然后用 process - (metaspace + overhead) 得出 total flink mem。再根据 total flink mem, 计算出各个子区域的大小。

让我们测试一下, conf/flink-conf.yaml:

# TaskManager settings
taskmanager.memory.process.size: 2g

计算流程如下:

JVM Metaspace: 默认是 256m
JVM Overhead: 默认是 process 内存的 0.1, 上下限: 1g/192m -> 2g * 0.1 = 204.8m
Total flink mem: 2g - 256m - 204.8m = 1587.2m
Off-heap:
1. Managed: 1587.2m * 0.4 (fraction) = 634.88m
2. Direct.Network: 1587.2m * 0.1 (fraction) 上下限: (1g/64m) = 158.72m
3. Direct.Framework: 128m (default)
4. Direct.Task: 0 (default)
Heap memory:
1. Framework: 128m (default)
2. Task: Total flink memory - Off-heap - Heap.Framework: 1587.2m - 921.6m - 128m = 537.6m

启动 tm 后，观察日志:

INFO  [] - Final TaskExecutor Memory configuration:
INFO  [] -   Total Process Memory:          2.000gb (2147483648 bytes)
INFO  [] -     Total Flink Memory:          1.550gb (1664299824 bytes)
INFO  [] -       Total JVM Heap Memory:     665.600mb (697932173 bytes)
INFO  [] -         Framework:               128.000mb (134217728 bytes)
INFO  [] -         Task:                    537.600mb (563714445 bytes)
INFO  [] -       Total Off-heap Memory:     921.600mb (966367651 bytes)
INFO  [] -         Managed:                 634.880mb (665719939 bytes)
INFO  [] -         Total JVM Direct Memory: 286.720mb (300647712 bytes)
INFO  [] -           Framework:             128.000mb (134217728 bytes)
INFO  [] -           Task:                  0 bytes
INFO  [] -           Network:               158.720mb (166429984 bytes)
INFO  [] -     JVM Metaspace:               256.000mb (268435456 bytes)
INFO  [] -     JVM Overhead:                204.800mb (214748368 bytes)

查看 tm jvm flags:

> jcmd {TM_PID} VM.flags
...
-XX:InitialHeapSize=704643072
-XX:MaxHeapSize=704643072

-XX:MaxDirectMemorySize=300647712
-XX:MaxMetaspaceSize=268435456
...

通过对比可以看到彼此之间的关系:

JVM Flag	Flink Component
-XX:MaxHeapSize	Total JVM Heap Memory(Framework + Task)
-XX:MaxDirectMemorySize	Total JVM Direct Memory(Framework + Task + Network)
-XX:MaxMetaspaceSize	JVM Metaspace

小结: tm 的内存主要花在 Managed 和 Heap。

JobManager

jm 的内存模型如下, 和 tm 比较简单了很多

JVM Heap: JobManager 的 JVM 堆内存。
Off-Heap: JobManager 的堆外内存（直接内存或本地内存）。
JVM Metaspace: Flink JVM 进程的 Metaspace。
JVM Overhead: JVM 开销, 用于其他 JVM 开销的本地内存，例如栈空间、垃圾回收空间等。

其中 Heap 主要用于框架和 checkpoint 回调时用户代码执行。

对于 Off-heap，默认配置下，Flink JobManager 没有限制 Off-heap 的内存使用，通过: jobmanager.memory.enable-jvm-direct-memory-limit: true 可以对 jm jvm 进程增加 -XX:MaxDirectMemorySize 从而限制 Off-heap 内存使用。

jm 的内存配置和 tm 一样，遵循 3 中模式规则, conf/flink-conf.yaml:

jobmanager.memory.process.size: 1g

计算规则和 tm 类似:

JVM Metaspace: 默认是 256m
JVM Overhead: 默认是 process 内存的 0.1, 上下限: 1g/192m -> 1024g * 0.1 = 102.4m, 由于低于最小值, 用 192m
JVM Off-heap: 默认是 128m
JVM Heap: 1024m - 256m - 192m - 128m = 448m

观察 jm 的启动日志:

INFO  [] - Final Master Memory configuration:
INFO  [] -   Total Process Memory: 1024.000mb (1073741824 bytes)
INFO  [] -     Total Flink Memory: 576.000mb (603979776 bytes)
INFO  [] -       JVM Heap:         448.000mb (469762048 bytes)
INFO  [] -       Off-heap:         128.000mb (134217728 bytes)
INFO  [] -     JVM Metaspace:      256.000mb (268435456 bytes)
INFO  [] -     JVM Overhead:       192.000mb (201326592 bytes)

观察 jm 的 vm flags:

> jcmd {SESSION_CLUSTER_PID} VM.flags

-XX:InitialHeapSize=469762048
-XX:MaxHeapSize=469762048
...
-XX:MaxMetaspaceSize=268435456 

限制 direct-memory-limit 之后:

-Xmx469762048
-Xms469762048
-XX:MaxDirectMemorySize=134217728
-XX:MaxMetaspaceSize=268435456

并不影响 jvm 其它区域的内存分配。

常见问题

官方文档列举了几个常见问题

OutOfMemoryError: Java heap space
OutOfMemoryError: Direct buffer memory
OutOfMemoryError: Metaspace
IOException: Insufficient number of network buffers

具体的解决途径参考: 内存常见问题(官方)

Arthas

Arthas 是一款 Java 应用调试的工具，功能强大，底层依赖 JVM Agent 机制。想要运行 Arthas 一般都需要完整的 JDK。依赖 JAVA_HOME/lib/tools.jar。

使用 K8S 部署时，我们的镜像一般会继承 Flink 官方的镜像，但是官方镜像的 JDK 并不是完整的 JDK。并没有诸如 jcmd, jstat 等调试程序，也没有 tools.jar。

为了激活(attach) Arthas 的 agent，需要使用 jattach。好在 jattach 在 ubuntu 上有 apt 包可以直接安装！

apt update && apt install jattach
jattach {JVM_PID} load instrument false /opt/arthas/arthas-agent.jar 

然后观察目标 JVM 进程是否监听了 3658 端口

netstat -anlp

最后启动 arthas-client.jar 连接 3658 进行调试。

java -jar arthas-client.jar

REFs

<<<EOF

在 macOS 上搭建 Iceberg

2024-02-23T00:00:00+00:00

眼过千遍，不如手过一遍。本文将描述如何在 macOS 上搭建 Iceberg。涉及到: Hadoop, Hive, Iceberg, Spark, Flink。

环境 & 版本

OS: macOS Ventura (M2)
Java: 1.8 (Amazon Corretto)
Hadoop(hdfs): 3.3.6
Hive: 2.3.9
Iceberg: 1.4.3
Spark: 3.5.0-hadoop3
Flink: 1.18 (1.17)

Hadoop

从 Hadoop 官网下载最新版，下载时注意对应 cpu 架构。

解压，配置环境变量:

export HADOOP_HOME=$HOME/opt/hadoop-3.3.6
export PATH=$HADOOP_HOME/bin:$HADOOP_HOME/sbin:$PATH

export HADOOP_CONFIG_DIR=$HADOOP_HOME/etc/hadoop
export HADOOP_CLASSPATH=`$HADOOP_HOME/bin/hadoop classpath`

export YARN_CONFIG_DIR=$HADOOP_HOME/etc/hadoop

在 core-site.xml configurations 下添加如下配置:

<property>
  <name>fs.defaultFS</name>
  <value>hdfs://localhost:9000</value>
</property>

在 hdfs-site.xml 中添加如下配置:

<property>
  <name>dfs.namenode.name.dir</name>
  <value>{YOUR_NAME_NODE_DIR}</value>
</property>

<property>
  <name>dfs.namenode.checkpoint.dir</name>
  <value>{YOUR_CHECKPOINT_DIR}</value>
</property>

<property>
  <name>dfs.datanode.data.dir</name>
  <value>{YOUR_DATA_NODE_DIR}</value>
</property>

<property>
  <name>dfs.replication</name>
  <value>1</value>
</property>

这 3 个目录默认都是基于 hadoop.tmp.dir，而 hadoop.tmp.dir 的默认值是 /tmp/hadoop-${user-name}。 tmp 目录会由于操作系统策略被不定期清空，导致 hdfs 出现数据一致性问题。

所以建议将 3 个目录设置到 tmp 之外。

初始化 namenode

hdfs namenode -format

启动

用 sbin/start-dfs.sh 启动 hdfs。用 jps -l | grep hadoop 查看进程是否启动:

org.apache.hadoop.hdfs.server.namenode.NameNode
org.apache.hadoop.hdfs.server.datanode.DataNode
org.apache.hadoop.hdfs.server.namenode.SecondaryNameNode

测试

hadoop fs -mkdir -p /tmp
hadoop fs -put README.txt /tmp
hadoop fs -cat /tmp/README.txt

Hive

下载，解压。

配置环境变量:

export HIVE_HOME=$HOME/opt/apache-hive-2.3.9-bin
export PATH=$HIVE_HOME/bin:$PATH

export HIVE_CONFIG_DIR=$HIVE_HOME/conf

为 hive 表配置 hdfs 目录:

hadoop fs -chmod g+w /tmp
hadoop fs -mkdir     /user/hive/warehouse
hadoop fs -chmod g+w /user/hive/warehouse

配置 hive, 在 conf 目录下增加 hive-site.xml 文件，增加如下配置:

<property>
  <name>system:java.io.tmpdir</name>
  <value>/tmp/hive/java</value>
</property>
<property>
  <name>system:user.name</name>
  <value>${user.name}</value>
</property>
<property>
  <name>javax.jdo.option.ConnectionURL</name>
  <value>jdbc:mysql://localhost:3306/hive_metastore</value>
</property>
<property>
  <name>javax.jdo.option.ConnectionDriverName</name>
  <value>com.mysql.cj.jdbc.Driver</value>
</property>
<property>
  <name>javax.jdo.option.ConnectionUserName</name>
  <value>root</value>
</property>
<property>
  <name>javax.jdo.option.ConnectionPassword</name>
  <value>{YOUR_MYSQL_PASSWORD}<value>
</property>

这里主要是配置 metastore，我们使用 mysql 作为 metastore db，默认是 derby 数据库。

配置完成后需要执行命令进行初始化:

bin/schematool -dbType mysql -initSchema

初始化成功后，可以查看下 mysql 的表是否正常，然后:

启动 hive server: bin/hiveserver2
启动 beeline client: bin/beeline -u jdbc:hive2://localhost:10000

CREATE TABLE invites (foo INT, bar STRING) PARTITIONED BY (ds STRING);

LOAD DATA LOCAL INPATH './examples/files/kv2.txt' OVERWRITE INTO TABLE invites PARTITION (ds='2008-08-15');

SELECT * FROM invites LIMIT 5;

Issues

如果在 beeline client 连接 hiveserver 时出现: User: XXX is not allowed to impersonate anonymous (state=,code=0) 错误，需要在 Hadoop core-site.xml 添加如下配置:

<property>
  <name>hadoop.proxyuser.{YOUR_USER_NAME}.groups</name>
  <value>*</value>
</property>
<property>
  <name>hadoop.proxyuser.{YOUR_USER_NAME}.hosts</name>
  <value>*</value>
</property>

其中 YOUR_USER_NAME 就是启动 hiveserver2 进程的用户，可以通过 whoami 获取。

这是因为 hive 默认会做一个冒充的行为: 冒充当前提交查询的这个用户来提交 hadoop mapred 任务。而一个用户允许被哪些人冒充，需要在 hadoop 内进行配置。

详细的解释可以查看这里: Stack Overflow: Cannot connect to hive using beeline, user root cannot impersonate anonymous

Spark

下载，解压。

虽然 Spark 文档说需要把 hive-site.xml 复制到 conf 目录下，但是我在实际测试过程中发现 Spark 会根据 HIVE_HOME 找到 hive 的 config，从而解析出 metastore。

所以只复制 core-site.xml 和 hdfs-site.xml 即可。

测试

CREATE TABLE IF NOT EXISTS src (key INT, value STRING) USING hive;

LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src;

SELECT * FROM src LIMIT 5;

Flink

下载，解压。

修改配置 conf/flink-conf.yaml:

jobmanager.memory.process.size: 1600m

taskmanager.resource.id: static-tm-res-id
taskmanager.memory.process.size: 1728m
taskmanager.numberOfTaskSlots: 10

固化 resource id，调整 JM 和 TM 内存。

启动: bin/start-cluster.sh

访问 webUI 查看是否启动成功: http://localhost:9090/#/overview

测试

启动 bin/sql-client.sh

CREATE TABLE employees (
  name VARCHAR,
  salary INT
) WITH ( 
  'connector' = 'filesystem',
  'path' = '{SPARK_HOME}/examples/src/main/resources/employees.json',
  'format' = 'json'
);

SELECT * FROM employees;

Issue

在启动 sql-client.sh 的时候，可能会遇到: java.lang.NoClassDefFoundError

Exception in thread "main" org.apache.flink.table.client.SqlClientException: Could not read from command line.
  at org.apache.flink.table.client.cli.CliClient.getAndExecuteStatements(CliClient.java:221)
  at org.apache.flink.table.client.cli.CliClient.executeInteractive(CliClient.java:179)
  at org.apache.flink.table.client.cli.CliClient.executeInInteractiveMode(CliClient.java:121)
  at org.apache.flink.table.client.cli.CliClient.executeInInteractiveMode(CliClient.java:114)
  at org.apache.flink.table.client.SqlClient.openCli(SqlClient.java:169)
  at org.apache.flink.table.client.SqlClient.start(SqlClient.java:118)
  at org.apache.flink.table.client.SqlClient.startClient(SqlClient.java:228)
  at org.apache.flink.table.client.SqlClient.main(SqlClient.java:179)
Caused by: java.lang.NoClassDefFoundError: Could not initialize class org.apache.flink.table.client.config.SqlClientOptions
  at org.apache.flink.table.client.cli.parser.SqlClientSyntaxHighlighter.highlight(SqlClientSyntaxHighlighter.java:59)
  at org.jline.reader.impl.LineReaderImpl.getHighlightedBuffer(LineReaderImpl.java:3633)
  at org.jline.reader.impl.LineReaderImpl.getDisplayedBufferWithPrompts(LineReaderImpl.java:3615)
  at org.jline.reader.impl.LineReaderImpl.redisplay(LineReaderImpl.java:3554)
  at org.jline.reader.impl.LineReaderImpl.doCleanup(LineReaderImpl.java:2340)
  at org.jline.reader.impl.LineReaderImpl.cleanup(LineReaderImpl.java:2332)
  at org.jline.reader.impl.LineReaderImpl.readLine(LineReaderImpl.java:626)
  at org.apache.flink.table.client.cli.CliClient.getAndExecuteStatements(CliClient.java:194)
  ... 7 more

这是因为 Flink 为了集成 Hadoop，会将 hadoop classpath 作为 classpath 一部分，查看 sql-client.sh 第 93 行:

exec "$JAVA_RUN" $FLINK_ENV_JAVA_OPTS $JVM_ARGS "${log_setting[@]}" -classpath "`manglePathList "$CC_CLASSPATH:$INTERNAL_HADOOP_CLASSPATHS:$FLINK_SQL_CLIENT_JAR"`" org.apache.flink.table.client.SqlClient "$@" --jar "`manglePath $FLINK_SQL_CLIENT_JAR`" 

由于 hadoop classpath 包含很多 jar，这些 jar 容易与 flink 依赖的 jar 产生版本冲突，所以解决方式是让 flink 的 jar 优先加载。调整 classpath 的顺序:

exec "$JAVA_RUN" $FLINK_ENV_JAVA_OPTS $JVM_ARGS "${log_setting[@]}" -classpath "`manglePathList "$CC_CLASSPATH:$FLINK_SQL_CLIENT_JAR:$INTERNAL_HADOOP_CLASSPATHS"`" org.apache.flink.table.client.SqlClient "$@" --jar "`manglePath $FLINK_SQL_CLIENT_JAR`" 

类似的错误可以参考这个: Issue(FLINK-33358): Flink SQL Client fails to start in Flink on YARN。

Flink 读取 Hive 表

不同于 Spark，Flink 查询 Hive 表时需要通过 Hive Thrift MetaServer。

启用 hive metastore server: hive --service metastore
在 {YOUR_FLINK_INSTALL_DIR}/conf 下新建 hive-site.xml: 注意此 hive-site.xml 是提供给 Flink 访问 hive metastore 使用的，不是 hive 安装目录下的 hive-site.xml。需要配置 metastore server 的地址和 hive dir 即可:
```
<property>
  <name>hive.metastore.uris</name>
  <value>thrift://localhost:9083</value>
</property>
<property>
  <name>hive.metastore.warehouse.dir</name>
  <value>/user/hive/warehouse</value>
</property>
```
将 flink-sql-connector-hive-2.3.9_2.12-1.18.1.jar 添加到 $FLINK_HOME/lib 下。重启 flink cluster。

启动 sql-client.sh，增加 hive catalog:

CREATE CATALOG myhive WITH (
  'type' = 'hive',
  'hive-conf-dir' = '{YOUR_FLINK_INSTALL_DIR}/conf'
);

Query:

Flink SQL> use catalog myhive;
[INFO] Execute statement succeed.

Flink SQL> show tables;
+-------------+
|  table name |
+-------------+
|     invites |
|         src |
+-------------+
4 rows in set
   
Flink SQL> select * from src limit 5;

Iceberg

准备工作终于结束！接下来让我们回到 Iceberg 上。

Spark 访问 Iceberg

下载 iceberg-spark-runtime-3.5_2.12:1.4.3.jar 并添加到 $SPARK_HOME/lib 下。

我们使用 spark-sql 来访问 Iceberg。启动:

spark-sql \
  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
  --conf spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog \
  --conf spark.sql.catalog.spark_catalog.type=hive \
  --conf spark.sql.catalog.local=org.apache.iceberg.spark.SparkCatalog \
  --conf spark.sql.catalog.local.type=hadoop \
  --conf spark.sql.catalog.local.warehouse=/user/iceberg/warehouse

然后直接按照文档进行 DDL、DML、Query 操作就好:

CREATE TABLE local.db.table (id bigint, data string) USING iceberg;

INSERT INTO local.db.table VALUES (1, 'a'), (2, 'b'), (3, 'c');

SELECT count(1) as count, data
FROM local.db.table
GROUP BY data;

CREATE TABLE spark_catalog.default.invites_ice(foo INT, bar STRING) PARTITIONED BY (ds STRING) USING iceberg;

INSERT INTO spark_catalog.default.invites_ice SELECT * FROM spark_catalog.default.invites;

值得一提的是，Spark 的 catalog 是懒加载的。这意味如果你第一次进入到 spark-sql 的环境，直接用 show catalogs 是无法查看到 local 的。

必须先先引用其中的 db 或者 table。

Flink 访问 Iceberg

我们先将下载 iceberg-flink-runtime-1.17-1.4.3.jar 注意: 由于 Flink 1.18 版本比较新，目前 Iceberg 还没有适配。

我偷个懒，直接用 1.17 的包试验下。

启动 sql-client:

Flink SQL> CREATE CATALOG `local` WITH (
  'type'='iceberg',
  'catalog-type'='hadoop',
  'warehouse'='/user/iceberg/warehouse'
);
[INFO] Execute statement succeed.

Flink SQL> use catalog `local`;
[INFO] Execute statement succeed.

Flink SQL> select * from `table`;

再来试下使用 hive metastore 的 Iceberg 表:

Flink SQL> CREATE CATALOG hive_catalog WITH (
  'type'='iceberg',
  'catalog-type'='hive',
  'uri'='thrift://localhost:9083',
  'clients'='5',
  'hive-conf-dir'='{YOUR_HIVE_CONFIG_DIR}',
  'hadoop-conf-dir'='{YOUR_HADOOP_CONFIG_DIR}'
);
[INFO] Execute statement succeed.

Flink SQL> use `default`;
[INFO] Execute statement succeed.

Flink SQL> show tables;
+-------------+
|  table name |
+-------------+
| invites_ice |
+-------------+

Flink SQL> select * from invites_ice;

终于，我们可以使用 Spark 和 Flink 正常操作 Iceberg 表。

REF

<<<EOF

Maven & Gradle 依赖作用域

2024-01-28T00:00:00+00:00

Maven 和 Gradle 是目前 Java(JVM) 项目中比较主流的构建工具，今天我们来聊一下他们各自如何管理依赖的作用域。

正确地理解依赖作用域对于我们开发出简洁、轻量的程序很有帮助。

在开始讲依赖作用域之前，我们先回忆一下 Java 的开发过程。

我们编写的代码，需要经过编译，变成字节码，然后以 jar 的形式交给 jvm 去运行。

所以，依赖作用域配置主要影响它在两个地方出现，一个是编译期(compile time)，一个是运行期(runtime)。

Maven Dependency Scope

Maven 的作用域比较直接，直接在 dependency 元素下声明 scope 即可。

<dependency>
  <groupId>com.alibaba</groupId>
  <artifactId>fastjson</artifactId>
  <version>1.17.25</version>
  <scope>compile</scope>
</dependency>

默认如果不配置，则是 compile.

`compile`

即在编译环节就需要引入的依赖，如果应用最终会打成 uber jar，那么这些依赖都会被打包到 uber jar 中。这也是 maven 里用得最多的作用域。

`provided` & `runtime`

这俩兄弟有一个共同的特点，就是不会把依赖带到 uber jar 中。

区别是: 声明为 provided 的依赖，Maven 会加入到编译期的 classpath 中，而 runtime 则不会。举个栗子:

比如我们做的是一个 web 项目，最终会部署到 ecs 上的 tomcat 容器里运行，那么项目里声明的 servlet-api 依赖，应该是 provided:

<dependency>
  <groupId>javax.servlet</groupId>
  <artifactId>javax.servlet-api</artifactId>
  <version>4.0.1</version>
  <scope>provided</scope>
</dependency>

为什么？程序在运行期，servlet-api 由容器统一管理，不需要单独打包到应用的 jar 包里。而我们在开发阶段，在编译期，开发功能时需要实现 servlet-api 相关的接口(import servlet-api 的类)，所以需要它在 classpath 上才能通过编译。

而 runtime 是彻底将依赖和应用 “断绝关系” ，编译期无法感知到它的存在，运行期由容器动态提供。一个例子是 slf4j:

项目在使用 slf4j 的时候，在编译期只需要引入 slf4j-api, 在程序运行期由具体的 classpath 上的 binding jar 来决定是使用 log4j or logback。

<dependency>
  <groupId>org.slf4j</groupId>
  <artifactId>slf4j-api</artifactId>
  <scope>compile</scope>
</dependency>
<dependency>
  <groupId>org.apache.logging.log4j</groupId>
  <artifactId>log4j-core</artifactId>
  <scope>runtime</scope>
</dependency>
<dependency>
  <groupId>org.apache.logging.log4j</groupId>
  <artifactId>log4j-slf4j2-impl</artifactId>
  <scope>runtime</scope>
</dependency>

`import`

这是一个特殊的 scope，只能应用在 type 为 pom 的依赖上，很简单，把该 pom 的 dependencyManagement 内容引入到当前的项目。

但是，为了减少外部依赖对项目的影响，建议尽量少使用。

为什么这么说呢？可以试想一下，假如我 import 的 pom 依赖中有某个 jar 的版本是 A，但是项目的传递依赖的版本是 B。就导致冲突了。

而往往 import 的 pom 中包含很多依赖，所以非常容易导致冲突。

`test`

被声明为 test 的依赖只出现在 test classpath 上。

Gradle

Gradle 首先明确出 compileClasspath 和 runtimeClasspath 两个概念。

可以这个命令查看依赖树的情况:

> ./gradlew dependencies

Gradle 的 dependency scope 除了影响依赖是否加入到对应的 classpath 中，还会决定是否将依赖暴露给第三方，这个特性为 lib 开发带来了方便。

Scope	When available	Leaks to consumer
implementation	compiletime + runtime	runtime
api	compiletime + runtime	compiletime + runtime
compiletimeOnly	compiletime	-
runtimeOnly	runtime	runtime

注: 这里的 consumer 是指 jar 包的消费者，也就是 jar 包的使用方。

回到刚才 Maven 的例子:

如果是 servlet-api，在 Gradle 中对应的是 compiletimeOnly。
如果是 slf4j, 对于 slf4j-api 我们需要声明为 implemtation(或者 api)，而 binding jar: log4j 或者 logback 则需要声明为 runtimeOnly。

除此之外，Gradle 还提供了 testApi, testImplemention 选项用于控制 test 依赖是否暴露给 consumer。

<<<EOF

在 Flink 任务中使用动态编译

2023-06-12T00:00:00+00:00

Java 动态编译是一项比较成熟的技术，用于在程序运行期间生成 Class ，为实现业务带来了极大的灵活性。本文讨论如何在 Flink 任务里使用动态编译，以及遇到的坑。

动态编译

在开始讲动态编译之前，先回忆一下静态编译。

静态编译

编写 .java 源代码
利用 javac 将源代码编译为 class 文件
打成可执行 jar 包发布

深入一点上述第 2 步的编译过程，假设现在有一个 HelloWorld.java 代码如下:

public class HelloWorld {
  public static void main(String[] args) {
    System.out.println("Hello World!");
  }
}

我们现在尝试通过 javac 来编译它, 打开 verbose 日志:

# 已格式化
> javac -verbose HelloWorld.java
[parsing started RegularFileObject[HelloWorld.java]]
[parsing completed 11ms]
[search path for source files: .]
[search path for class files: 
  ${JDK_HOME}/Contents/Home/jre/lib/resources.jar,
  ${JDK_HOME}/Contents/Home/jre/lib/rt.jar,
  ${JDK_HOME}/Contents/Home/jre/lib/sunrsasign.jar,
  ${JDK_HOME}/Contents/Home/jre/lib/jsse.jar,
  ${JDK_HOME}/Contents/Home/jre/lib/jce.jar,
  ${JDK_HOME}/Contents/Home/jre/lib/charsets.jar,
  ${JDK_HOME}/Contents/Home/jre/lib/jfr.jar,
  ${JDK_HOME}/Contents/Home/jre/classes,
  ${JDK_HOME}/Contents/Home/jre/lib/ext/sunec.jar,
  ${JDK_HOME}/Contents/Home/jre/lib/ext/nashorn.jar,
  ${JDK_HOME}/Contents/Home/jre/lib/ext/cldrdata.jar,
  ${JDK_HOME}/Contents/Home/jre/lib/ext/jfxrt.jar,
  ${JDK_HOME}/Contents/Home/jre/lib/ext/dnsns.jar,
  ${JDK_HOME}/Contents/Home/jre/lib/ext/localedata.jar,
  ${JDK_HOME}/Contents/Home/jre/lib/ext/sunjce_provider.jar,
  ${JDK_HOME}/Contents/Home/jre/lib/ext/sunpkcs11.jar,
  ${JDK_HOME}/Contents/Home/jre/lib/ext/jaccess.jar,
  ${JDK_HOME}/Contents/Home/jre/lib/ext/zipfs.jar,
  .
]
[loading ZipFileIndexFileObject[${JDK_HOME}/Contents/Home/lib/ct.sym(META-INF/sym/rt.jar)]:
  java/lang/Object.class, 
  java/lang/String.class
]
[checking HelloWorld]
[loading ZipFileIndexFileObject[${JDK_HOME}/Contents/Home/lib/ct.sym(META-INF/sym/rt.jar)]:
  java/io/Serializable.class,
  java/lang/AutoCloseable.class,
  java/lang/Byte.class,
  java/lang/Character.class,
  java/lang/Short.class,
  java/lang/Long.class,
  java/lang/Float.class,
  java/lang/Integer.class,
  java/lang/Double.class,
  java/lang/Boolean.class,
  java/lang/Void.class,
  java/lang/System.class,
  java/io/PrintStream.class,
  java/lang/Appendable.class,
  java/io/Closeable.class,
  java/io/FilterOutputStream.class,
  java/io/OutputStream.class,
  java/io/Flushable.class,
  java/lang/Comparable.class,
  java/lang/CharSequence.class
]
[wrote RegularFileObject[HelloWorld.class]]
[total 277ms]

可以看到，javac 在编译的过程中使用到了 classpath, 其包含 jdk 内置的运行时 jar 包目录, class 目录, ext 目录, 以及用户代码的 classpath: (默认是 .)

编译的过程中，有两次加载 class 发生。分别是在 [checking HelloWorld] 之前和之后。

可以推断，第一次是验证 HelloWorld.java 时，对其依赖的 class 做递归式地加载，第二次是实现编译过程时，对过程中依赖的 class 加载。所以才看到有 io 相关的 class。

javac 对 rt.jar 下的 class 做了优化, 并没有直接引用，而是通过 ct.sym 去加载。

最后，将编译好的 class 写到文件: RegularFileObject[HelloWorld.class]

动态编译

动态编译的过程和静态编译类似, 带入代码我们来看下。

public static void compile(String className, String source, String outputDir) {
  // JavaSourceFile: 实现了 JavaFileObject 的 source 实现.
  JavaSourceFile sourceFile = new JavaSourceFile(className, source);

  // 获取 compiler 实例
  JavaCompiler compiler = ToolProvider.getSystemJavaCompiler();
  StandardJavaFileManager stdFileManager = compiler.getStandardFileManager(null, null, null);

  // 拦截 fileManager 方法以指定 input 和 output
  JavaFileManager fileManager = new ForwardingJavaFileManager<StandardJavaFileManager>(stdFileManager) {
    @Override
    public JavaFileObject getJavaFileForInput(
      Location location, String className, JavaFileObject.Kind kind
    ) throws IOException {
      return sourceFile;
    }
  };
  
  // 编译诊断收集器
  DiagnosticCollector<JavaFileObject> collector = new DiagnosticCollector<>();

  // 构造编译任务
  JavaCompiler.CompilationTask task = compiler.getTask(
      new OutputStreamWriter(System.err),
      fileManager,
      collector,
      Arrays.asList("-d", outputDir),
      null,
      Collections.singleton(sourceFile)
  );

  Boolean success = task.call();
  if (!success) {
    collector.getDiagnostics().forEach(System.err::println);
  } else {
    System.out.println("Compile success");
  }
}

首先构造编译的原材料: 源代码 JavaFileObject, 然后向 jvm 获取 JavaCompiler 实例, 通过继承 JavaCompiler 下的 JavaFileManager 拦截了 getJavaFileForInput 方法。

现在要开始编译了: 在 sun 提供的 api 里，CompilationTask 就代表 shell 命令 javac HelloWorld.java。 getTask 接纳了 JavaFileManager, 编译诊断的收集器, 编译选项, 注解处理, 编译单元(source file)。

吐槽下 sun 提供的 api 其实并不优雅, fileManager 以及编译单元在 getTask 时感觉都可以省略。另外 fileManager 默认和磁盘文件耦合太深，而动态编译通常都是在内存完成的。

最后，如果 task.call 返回了 true 则表示编译成功。那么问题来了，编译生成的 class 在哪儿呢?

有好几种实现方式，可以拦截 fileManager 的 getJavaFileForOutput 实现，也可以在通过在编译选项上增加 -d 选项指明 class 的输出目录。个人认为后一种方式更灵活。

至此，动态编译完成。小结下:

动态编译的本质就是通过 java api 实现了在运行期执行 javac 命令这个过程。
api 依赖的 jdk 内置的 jar 包 (位于 ext 目录的 tools.jar )，jre 无法完成。

所以说动态编译实际上依赖了运行时 jdk，不同 jdk 厂商的 api 可能会有差异。

在实现的时候可以参考 github 上比较完善的实现:

Java-Runtime-Compiler: 开箱即用的企业级动态编译器, 支持缓存，
compiler: 编译过程在内存完成，仅支持 java6, 已停止维护。
Janino: drools 早期使用的动态编译器，功能强大，api 抽象层次较低，上手难度大。

在 Operator 实现

编译阶段

为了避免动态编译的 api 和 jdk 耦合，做的第一件事情就是将 java tools 这个 jar 包打进 fat jar 内部。

添加 dependency:

<dependency>
  <groupId>com.perfma.wrapped</groupId>
  <artifactId>com.sun.tools</artifactId>
  <version>1.8.0_jdk8u275-b01_linux_x64</version>
</dependency>

在 compiler 解析 source code (A) 的时候，如果发现依赖了另外一些 class (B) 时，会尝试 load B。而 B 往往是用户代码 fat jar 的内容。

如果 fat jar 已经在当前 jvm 的 classpath 上了。那肯定可以找到，问题在于不是。

当 Flink Cluster 启动的时候，用户的 jar 还没有上传。Flink Cluster 启动了 JobManager 和 TaskManager 初始化对应的 Slots。

以 TaskManager 为例，其 classpath 可以在启动日志中看到:

2023-06-19 11:24:43,523 [main] INFO  org.apache.flink.kubernetes.taskmanager.KubernetesTaskExecutorRunner [] -  Classpath: 
/flink/lib/flink-cep-1.15-vvr-6.0.6-SNAPSHOT.jar:
/flink/lib/flink-connector-files-1.15-vvr-6.0.6-SNAPSHOT.jar:
/flink/lib/flink-csv-1.15-vvr-6.0.6-SNAPSHOT.jar:
/flink/lib/flink-json-1.15-vvr-6.0.6-SNAPSHOT.jar:
/flink/lib/flink-metrics-log-1.15-vvr-6.0.6-SNAPSHOT.jar:
/flink/lib/flink-metrics-prometheus-1.15-vvr-6.0.6-SNAPSHOT.jar:
/flink/lib/flink-metrics-slf4j-1.15-vvr-6.0.6-SNAPSHOT.jar:
/flink/lib/flink-ml-examples-2.1-1.0.0-SNAPSHOT.jar:
/flink/lib/flink-ml-uber-2.1-1.0.0-SNAPSHOT.jar:
/flink/lib/flink-queryable-state-runtime_*.jar:
/flink/lib/flink-scala_2.12-1.15-vvr-6.0.6-SNAPSHOT.jar:
/flink/lib/flink-shaded-hadoop-2-uber-3.1.3-10.0-SNAPSHOT.jar:
/flink/lib/flink-shaded-zookeeper-3.5.9.jar:
/flink/lib/flink-statebackend-gemini_1.8-3.2.6-SNAPSHOT.jar:
/flink/lib/flink-state-processor-api-1.15-vvr-6.0.6-SNAPSHOT.jar:
/flink/lib/flink-table-api-java-uber-1.15-vvr-6.0.6-SNAPSHOT.jar:
/flink/lib/flink-table-planner-loader-1.15-vvr-6.0.6-SNAPSHOT.jar:
/flink/lib/flink-table-runtime-1.15-vvr-6.0.6-SNAPSHOT.jar:
/flink/lib/jersey-core-1.9.jar:
/flink/lib/log4j-1.2-api-2.17.1.jar:
/flink/lib/log4j-api-2.17.1.jar:
/flink/lib/log4j-core-2.17.1.jar:
/flink/lib/log4j-slf4j-impl-2.17.1.jar:
/flink/lib/flink-dist-1.15-vvr-6.0.6-SNAPSHOT.jar:
:
:

classpath 只包含 Flink 框架的 jar。用户提交的 fat jar 由 Flink 自定义的 ClassLoader (FlinkUserCodeClassLoader) 进行加载。

当用户通过 Flink 客户端将 Job 提交到 Cluster, JobManager 会将 fat jar 注册到 BlobService 并缓存到本地。 TaskManager 通过 rpc 向 JobManager 下载 fat jar 到本地，然后通过自定义的 ClassLoader 加载 jar 包，开始运行 StreamTask。

所以，在动态编译期，我们需要显式地指定 classpath，相当于告诉 javac 当前 fat jar 的路径。

List<String> options = new ArrayList<>();
options.add("-classpath");
StringBuilder sb = new StringBuilder();

// 通过 FlinkUserCodeClassLoader 获取 fat jar 路径
URLClassLoader urlClassLoader = (URLClassLoader) this.getClass().getClassLoader();

for (URL url : urlClassLoader.getURLs()) {
  sb.append(url.getFile()).append(System.getProperty("path.separator"));
}

options.add(sb.toString());
log.info("FlinkUserCodeClassLoader classpath: {}", sb);

JavaCompiler.CompilationTask task = compiler.getTask(null, manager, collector, options, null, Collections.singletonList(source));

加载阶段

编译完成后，想要使用这个 class (A), 需要通过 ClassLoader 加载它。可以直接通过 FlinkUserCodeClassLoader，前提是将编译完成的 class 置于其 classpath 路径中。

也可以自定义一个 ClassLoader，但是需要注意将它的 parent 设置为 FlinkUserCodeClassLoader。为什么？

因为如果不指定 parent ，那么通过 ClassLoader 源码可知，将会默认继承 SystemClassLoader 作为 parent:

// java.lang.ClassLoader.java: 
protected ClassLoader() {
  this(checkCreateClassLoader(), getSystemClassLoader());
}

而刚才我们看了, SystemClassLoader 的 classpath 只包含 Flink 框架的类库，所以不指定 parent 时, 是无法加载 user code 里的 class 的。 load 时会报: ClassNotFound。

最后用一张图来说明动态编译类的加载:

<<<EOF

使用 rync 同步本地和远程的目录

2023-06-12T00:00:00+00:00

如果你有一个超大的文件夹需要频繁在本地和远程同步，简单的 scp 由于每次是全量复制，会导致产生很多无用功。

增量同步工具 rsync 登场~ 🔫

原文链接

介绍

Rsync 含义是 “Remote Sync”, 是 linux 下用来同步远程和本地目录工具。它使用特殊的算法，来计算由于变化而需要复制的数据，使传输数据量最小化，即通过算法实现文件的增量同步。

在这篇文章中，我们将介绍这强大工具的基本使用。我们使用 Ubuntu 12.04 vps，但是你可以在任何 linux 发行版上执行命令。

什么是 rsync

rsync 是一款非常灵活的基于网络同步工具。它同样可以指代工具内部使用的 rsync 网络协议。

约定: 当我们在文章中说到 rsync 时，我们指的是工具而非协议。

由于它在 Linux 和类 Unix 系统上无处不在，并且作为系统脚本工具的流行，大多数 Linux 发行版默认包含它。

基本使用

rsync 的基本语法非常直接，就像是在使用 ssh, scp, cp.

我们先创建两个目录和若干测试文件:

cd ~
mkdir dir1
mkdir dir2
touch dir1/file{1..100}

现在我们有一个名为 dir1 的目录，它包含了 100 个空文件:

> ls dir1
file1    file18  file27  file36  file45  file54  file63  file72  file81  file90
file10   file19  file28  file37  file46  file55  file64  file73  file82  file91
file100  file2   file29  file38  file47  file56  file65  file74  file83  file92
file11   file20  file3   file39  file48  file57  file66  file75  file84  file93
file12   file21  file30  file4   file49  file58  file67  file76  file85  file94
file13   file22  file31  file40  file5   file59  file68  file77  file86  file95
file14   file23  file32  file41  file50  file6   file69  file78  file87  file96
file15   file24  file33  file42  file51  file60  file7   file79  file88  file97
file16   file25  file34  file43  file52  file61  file70  file8   file89  file98
file17   file26  file35  file44  file53  file62  file71  file80  file9   file99

我们还创建了一个 dir2 的空目录。在本地系统上，将 dir1 的内容同步到 dir2, 可以用:

> rsync -r dir1/ dir2

-r 选项代表递归 (recursive)，在同步目录时一定要带上。也可以附上 -a 选项:

> rsync -a dir1/ dir2

-a 是一个复合选项，a 代表 archive, 递归同步，并且保留链接, 特殊设备文件, 修改时间, 用户组，文件所属用户，权限. 比起 -r , -a 的使用频率更高。

注意

你也许已经注意到第一个参数末尾带上了斜杠(/)

> rsync -a dir1/ dir2

这个斜杠代表 dir1 下的所有内容，不可省略。

如果不小心忘记了斜杠，rsync 将 dir1 连同内容放到 dir2 下，dir2 的目录结构将会变成:

~/dir2/dir1/[files]

在执行 rsync 命令之前，一定别忘记再检查一遍你的命令。

rsync 提供一种检查命令的方式: -n 或 –dry-run 选项。 -v (verbose) 选项会打印详细输出。

> rsync -anv dir1/ dir2

sending incremental file list
./
file1
file10
file100
file11
file12
file13
file14
file15
file16
file17
file18
...

对比一下我们去掉斜杠后 rsync 的操作有何不同:

> rsync -anv dir1 dir2
sending incremental file list
dir1/
dir1/file1
dir1/file10
dir1/file100
dir1/file11
dir1/file12
dir1/file13
dir1/file14
dir1/file15
dir1/file16
dir1/file17
dir1/file18
...

你可以看到目录本身也被传输了。

如何用 rsync 在本地和远程的系统间实现同步

如果你能 ssh 到远程机器，并且在本地和远程同时安装了 rsync, 那么进行同步非常简单。如果你需要设置 ssh keys 可以参考这篇文章。

一旦配置好了两台机器的 ssh ，你可以按如下命令将刚才的 dir1 同步到远程机器上 (注意我现在想要传输整个目录，所以我忽略了末尾的斜杠):

> rsync -a ~/dir1 username@remote_host:destination_directory

这也可以称为 推送(push) 操作，因为我把本地的一个目录推送到了远程机器上。

相反的操作成为拉取(pull)。意味着将远程机器的目录同步到本地。如果 dir1 在远程机器，语法会是这样:

> rsync -a username@remote_host:/home/username/place_to_sync_on_local_machine

就像 cp 一样，源文件永远位于第一个参数，目的文件在第二位。

rsync 的重要选项

rsync 为修改默认行为提供了丰富的选项。我们之前已经讨论过了一部分。

如果你想传输的文件是未被压缩过的文本文件，你可以通过 -z 压缩选项来减少网络传输:

> rsync -az source destination

-P 标记非常有用，它是 --progress 和 --partial 的结合。--progress 将会为你呈现传输进度条，--partial 允许你继续被中断的传输。

> rsync -azP source destination
sending incremental file list
./
file1
						0 100%    0.00kB/s    0:00:00 (xfer#1, to-check=99/101)
file10
						0 100%    0.00kB/s    0:00:00 (xfer#2, to-check=98/101)
file100
						0 100%    0.00kB/s    0:00:00 (xfer#3, to-check=97/101)
file11
						0 100%    0.00kB/s    0:00:00 (xfer#4, to-check=96/101)
...

如果你再次运行该命令，你会看到更简短的输出，因为 rsync 已经判断出文件没有修改发生，无需传输。

为了说明 rsync 可以通过文件修改时间来决定它们是否发生了变化:

> rsync -azP source destination
sending incremental file list

sent 818 bytes received 12 bytes 1660.00 bytes/sec
total size is 0 speedup is 0.00

我们可以更新几个文件的修改时间来查看 rsync 时如何智能地只传输变化的文件:

> touch dir1/file{1..10}
> rsync -azP source destination
sending incremental file list
file1
						0 100%    0.00kB/s    0:00:00 (xfer#1, to-check=99/101)
file10
						0 100%    0.00kB/s    0:00:00 (xfer#2, to-check=98/101)
file2
						0 100%    0.00kB/s    0:00:00 (xfer#3, to-check=87/101)
file3
						0 100%    0.00kB/s    0:00:00 (xfer#4, to-check=76/101)
...

为了让两个目录保持绝对同步，对于已经在源目录中被删除的文件，在目标目录里，有必要也删掉它们。默认情况下，rsync 不会从目标目录中删除任何文件。

我们可以通过增加 --delete 标记来修改这一行为，推荐使用 --dry-run 来测试命令以防止数据丢失。

> rsync -a `--delete` source destination

如果你想在同步时，从从源目录排除部分文件或者目录，可以在 --exclude= 选项后用一个逗号分隔的列表来指定它们:

> rsync -a --exclude=pattern_to_exclude source destination

推荐 –exclude 升级版: –exclude-file 可以将 pattern 记录在文件中, 通常 gitignore 的匹配规则同样适用 rsync 的 exclude。

如果我们已经在命令上制定了排除(exclude)的参数，仍可以通过制定包含(include)参数来强制同步:

> rsync -a --exclude=pattern_to_exclude --include=pattern_to_include source destination

小结

rsync 通过网络进行文件同步，简单、强大、灵活使它成为文件级别同步工具的不二之选。

掌握 rsync 可让您设计复杂的备份操作并获得对传输内容和传输方式的细粒度控制。

<<<EOF

Kwritin

Flink 概念辨析 Operator, Task, Subtask, Slot

创建 Demo

分析 Demo

Flink 内存配置实践

本地测试环境搭建

TaskManager

JobManager

常见问题

Arthas

REFs

在 macOS 上搭建 Iceberg

环境 & 版本

Hadoop

初始化 namenode

启动

测试

Hive

Issues

Spark

测试

Flink

测试

Issue

Flink 读取 Hive 表

Iceberg

Spark 访问 Iceberg

Flink 访问 Iceberg

REF

Maven & Gradle 依赖作用域

Maven Dependency Scope

compile

provided & runtime

import

test

Gradle

在 Flink 任务中使用动态编译

动态编译

静态编译

动态编译

在 Operator 实现

编译阶段

加载阶段

使用 rync 同步本地和远程的目录

介绍

什么是 rsync

基本使用

注意

如何用 rsync 在本地和远程的系统间实现同步

rsync 的重要选项

小结

`compile`

`provided` & `runtime`

`import`

`test`