标签: apache-flink

Apache Flink 通过从检查点恢复作业来保证故障时的一次性处理和恢复，检查点是分布式数据流和算子状态的一致快照（分布式快照的 Chandy-Lamport 算法）。 这保证了故障转移时恰好一次。

在集群正常运行的情况下，Flink 如何保证一次处理，例如给定一个从外部源（比如 Kafka）读取的 Flink 源，Flink 如何保证事件从源读取一次？事件源和 Flink 源之间是否存在任何类型的应用程序级别确认？另外，Flink 如何保证事件从上游算子到下游算子只传播一次？这是否也需要对收到的事件进行任何类型的确认？

当我需要使用 I/O（查询 DB、调用第三个 API，...）时，我可以使用 RichAsyncFunction。但我需要通过 GG Sheet API 与 Google Sheet 进行交互： https: //developers.google.com/sheets/api/quickstart/java。这个API是同步的。我写了下面的代码片段：

public class SendGGSheetFunction extends RichAsyncFunction<Obj, String> {

    @Override
    public void asyncInvoke(Obj message, final ResultFuture<String> resultFuture) {
        CompletableFuture.supplyAsync(() -> {
            syncSendToGGSheet(message);
            return "";
        }).thenAccept((String result) -> {
            resultFuture.complete(Collections.singleton(result));
        });
    }

}

但我发现消息发送到GGSheet非常慢，看起来是同步发送的。

示例代码：

from pyflink.table import EnvironmentSettings, StreamTableEnvironment


env_settings = (
    EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build()
)
table_env = StreamTableEnvironment.create(environment_settings=env_settings)

table_env.execute_sql(
    """
    CREATE TABLE table1 (
        id INT,
        ts TIMESTAMP(3),
        WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
    ) WITH (
        'connector.type' = 'filesystem',
        'format.type' = 'csv',
        'connector.path' = '/home/alex/work/test-flink/data1.csv'
    )
"""
)


table_env.execute_sql(
    """
    CREATE TABLE table2 (
        id2 INT,
        ts2 TIMESTAMP(3),
        WATERMARK FOR ts2 AS ts2 - INTERVAL '5' SECOND
    ) WITH (
        'connector.type' = 'filesystem',
        'format.type' = 'csv',
        'connector.path' = '/home/alex/work/test-flink/data2.csv'
    ) …

我的工作有状态操作员并且还启用了检查点。staful 操作员的任务之一由于某种原因失败，已重新启动并恢复检查点状态。

我想问以下哪一个是重启行为：

1. 仅重新启动并恢复失败的任务
2. 所有操作员（包含失败的任务）的任务都将重新启动并恢复
3. 整个作业重新启动并恢复

我需要延迟处理某些事件。

例如。我有三个事件（发布在 Kafka 上）：

• A（id：1，重试时间：现在）
• B（id：2，重试时间：10分钟后）
• C（id：3，重试时间：现在）

我需要立即处理记录A和C，而记录B需要十分钟后处理。在 Apache Flink 中这是否可行？

到目前为止，无论我研究过什么，“触发器”似乎可能有助于在 Flink 中实现它，但尚未能够正确实现它。

我也查看了 Kafka 文档，但看起来不太可行。

我正在按照Flink 官方教程在本机 Kubernetes 中启动会话。

首先，我创建了一个干净的新集群。

然而，运行后

./bin/kubernetes-session.sh -Dkubernetes.cluster-id=my-first-flink-cluster

my-first-flink-cluster-xxx我在刚刚创建的pod 日志中收到错误：

2021-08-14 18:33:02,519 WARN  io.fabric8.kubernetes.client.dsl.internal.WatchConnectionManager [] - Exec Failure: HTTP 403, Status: 403 - pods is forbidden: User "system:serviceaccount:default:default" cannot watch resource "pods" in API group "" in the namespace "default"
java.net.ProtocolException: Expected HTTP 101 response but was '403 Forbidden'
    at org.apache.flink.kubernetes.shaded.okhttp3.internal.ws.RealWebSocket.checkResponse(RealWebSocket.java:229) [flink-dist_2.12-1.13.1.jar:1.13.1]
    at org.apache.flink.kubernetes.shaded.okhttp3.internal.ws.RealWebSocket$2.onResponse(RealWebSocket.java:196) [flink-dist_2.12-1.13.1.jar:1.13.1]
    at org.apache.flink.kubernetes.shaded.okhttp3.RealCall$AsyncCall.execute(RealCall.java:206) [flink-dist_2.12-1.13.1.jar:1.13.1]
    at org.apache.flink.kubernetes.shaded.okhttp3.internal.NamedRunnable.run(NamedRunnable.java:32) [flink-dist_2.12-1.13.1.jar:1.13.1]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_302]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_302]
    at java.lang.Thread.run(Thread.java:748) [?:1.8.0_302]
2021-08-14 18:33:02,585 INFO  org.apache.flink.kubernetes.kubeclient.resources.KubernetesPodsWatcher [] - The …