基于SLS+Blink的实时计算最佳实践

日志服务简介

阿里云的日志服务（SLS）是针对日志类数据的一站式服务，无需开发就能快捷完成海量日志数据的采集、消费、投递以及查询分析等功能，提升运维、运营效率。在采集端支持30多种写入方式，包括自研的客户端Logtail，开源软件如Logstash、Fluent，Flume，Beats等，各种语言的SDK/Producer，无论是嵌入式设备、网页、服务器、程序等都能轻松接入。在消费端，支持与Storm、Spark Streaming、Flink/Blink等大数据系统无缝对接。

阿里云实时计算（Blink）

阿里云实时计算是基于Apache Flink构建的一站式、高性能实时大数据处理平台，广泛适用于流式数据处理、离线数据处理等场景。阿里云实时计算提供了如下两种数据处理API：

Flink SQL：通过DDL的方式定义Source和Sink，用SQL来实现数据的处理。
Flink Datastream: 在程序中使用各个Source和Sink的SDK，通过提交jar的方式运行托管的Flink Job。

通过这两种API，既可以把SLS作为数据源（Source），实现日志端到端的实时采集与处理，也可以把SLS作为结果的输出目标（Sink），在SLS对结果实时查询分析，配置可视化报表等。

Flink SQL

Flink SQL是阿里云实时计算为了简化计算模型、降低用户使用实时计算门槛而设计的一套符合标准SQL语义的开发语言。

创建SLS源表

SLS源表对应SLS中的Logstore，表中的字段与Logstore中日志字段一一映射，像执行SQL一样流式处理SLS中的数据。除了Logstore所属的Region对应的Endpoint和Project之外，还需要具有读SLS Logstore权限的Access Key以及消费数据起始位置对应的时间点。

create table sls_stream(
  a INT,
  b INT,
  c VARCHAR
) with (
  type ='sls',
  endPoint ='http://cn-hangzhou-share.log.aliyuncs.com',
  accessId ='<yourAccessId>',
  accessKey ='<yourAccessKey>',
  startTime = '2017-07-05 00:00:00',
  project ='<yourProjectName>',
  logStore ='<yourLogStoreName>',
  consumerGroup ='<yourConsumerGroupName>'
);

Checkpoint

SLS的底层存储Loghub，是一个类似Kafka的Append Only的存储系统，覆盖Kafka 100%的功能。与Kafka的partition类型，Logstore中的数据存储在每个Shard中。每个Shard都可以通过cursor或者时间戳，确定日志在Shard中的存储位置（对应Kafka中的Offset）。在消费过程中，为了支持程序重启时尽可能少的数据重复，需要将最新消费位置记录下来，用于进程重启之后继续消费。这个位置就是我们所说的checkpoint。

目前，Flink SQL任务checkpoint保存在Flink的State中，如果任务Failover 或者暂停之后恢复，会从State中恢复消费位置继续消费。然而如果任务重启或者其他原因造成的State丢失，任务将从启动时指定的时间点开始消费。创建源表时建议指定参数consumerGroup，Blink将自动同步消费位置到SLS服务端，以用于在SLS控制台监控消费进度。注意：任务重启时不会使用SLS服务端的消费位置恢复。

创建SLS结果表

通过定义结果表，通过INSERT语句，把从源表中处理之后的数据写入到SLS的Logstore中。

create table sls_output(
 `name` VARCHAR,
 age BIGINT,
 birthday BIGINT
)with(
 type='sls',
 endPoint='http://cn-hangzhou-corp.sls.aliyuncs.com',
 accessId='<yourAccessId>',
 accessKey='<yourAccessKey>',
 project='<yourProjectName>',
 logstore='<yourLogstoreName>'
);

INSERT INTO sls_output SELECT age, birthday FROM source_table;

Flink Datastream

Flink SQL 的不足

Flink SQL已经能够实现许多场景下的数据处理需求，但是受限于SQL的表达能力，对于比较复杂的业务场景，SQL实现起来比较复杂。
对于DDL 的定义，需要指定固定的字段列表，对于日志场景而言，日志字段不固定的情况非常普遍，这就意味着很难提前定义好全部的字段。
在日志中，很多时候单个字段是JSON或者其他复杂的形式，如果要在SQL里面解析和处理这类字段，不如在程序中之间处理起来灵活。

对于这些场景，可以考虑使用Data Stream API，通过自定义程序，实现更复杂的业务逻辑。

Maven 依赖

SLS开发了一个与开源Flink集成的SDK，同样适用于Blink。

<dependency>
    <groupId>com.aliyun.openservices</groupId>
    <artifactId>flink-log-connector</artifactId>
    <version>0.1.19.1</version>
</dependency>

Github源码：

https://github.com/aliyun/aliyun-log-flink-connector

消费SLS示例

public class ConsumerSample {
    private static final String SLS_ENDPOINT = "";
    private static final String ACCESS_KEY_ID = "";
    private static final String ACCESS_KEY_SECRET = "";
    private static final String SLS_PROJECT = "";
    private static final String SLS_LOGSTORE = "";

    public static void main(String[] args) throws Exception {
        final ParameterTool params = ParameterTool.fromArgs(args);
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        env.getConfig().setGlobalJobParameters(params);
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setStateBackend(new FsStateBackend("your checkpoint dir"));
        Properties configProps = new Properties();
        configProps.put(ConfigConstants.LOG_ENDPOINT, SLS_ENDPOINT);
        configProps.put(ConfigConstants.LOG_ACCESSSKEYID, ACCESS_KEY_ID);
        configProps.put(ConfigConstants.LOG_ACCESSKEY, ACCESS_KEY_SECRET);
        configProps.put(ConfigConstants.LOG_MAX_NUMBER_PER_FETCH, "10");
        configProps.put(ConfigConstants.LOG_CONSUMER_BEGIN_POSITION, Consts.LOG_FROM_CHECKPOINT);
        configProps.put(ConfigConstants.LOG_CONSUMERGROUP, "23_ots_sla_etl_product1");
        configProps.put(ConfigConstants.LOG_CHECKPOINT_MODE, CheckpointMode.ON_CHECKPOINTS.name());
        configProps.put(ConfigConstants.LOG_COMMIT_INTERVAL_MILLIS, "10000");

        FastLogGroupDeserializer deserializer = new FastLogGroupDeserializer();
        DataStream<FastLogGroupList> stream = env.addSource(
                new FlinkLogConsumer<>(SLS_PROJECT, SLS_LOGSTORE, deserializer, configProps));

        stream.flatMap((FlatMapFunction<FastLogGroupList, String>) (value, out) -> {
            for (FastLogGroup logGroup : value.getLogGroups()) {
                int logCount = logGroup.getLogsCount();
                for (int i = 0; i < logCount; i++) {
                    FastLog log = logGroup.getLogs(i);
                    // processing log
                }
            }
        });
        stream.writeAsText("log-" + System.nanoTime());
        env.execute("Flink consumer");
    }
}

与SQL类似，SDK也会把checkpoint保存到Flink State中，此外，还支持把Checkpoint同步到SLS服务端，这样当Flink本地的State无法恢复时，还能从服务端获取checkpoint，从而保证即便任务重启甚至重建，只要服务端的消费组没有删除，checkpoint就不会丢失。SDK同步checkpoint支持如下策略：

与Flink snapshotState同步，即在Flink调用snapshotState时，同步到服务端。默认是这种方式。
自动提交checkpoint，即定时提交checkpoint到服务端。好处是当Flink下游任务不支持exactly once时最大程度上避免数据重复。
不同步到SLS服务端。

处理结果写入SLS示例

public class ProducerSample {
    private static final String SLS_ENDPOINT = "cn-hangzhou.log.aliyuncs.com";
    private static final String ACCESS_KEY_ID = "";
    private static final String ACCESS_KEY = "";
    private static final String SLS_PROJECT = "";
    private static final String SLS_LOGSTORE = "test-flink-producer";

    public static void main(String[] args) throws Exception {
        final ParameterTool params = ParameterTool.fromArgs(args);
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().setGlobalJobParameters(params);
        env.setParallelism(3);
        DataStream<String> stream = env.addSource(new EventsGenerator());
        Properties configProps = new Properties();
        configProps.put(ConfigConstants.LOG_ENDPOINT, SLS_ENDPOINT);
        configProps.put(ConfigConstants.LOG_ACCESSSKEYID, ACCESS_KEY_ID);
        configProps.put(ConfigConstants.LOG_ACCESSKEY, ACCESS_KEY);
        configProps.put(ConfigConstants.LOG_PROJECT, SLS_PROJECT);
        configProps.put(ConfigConstants.LOG_LOGSTORE, SLS_LOGSTORE);

        FlinkLogProducer<String> logProducer = new FlinkLogProducer<>(new SimpleLogSerializer(), configProps);
        logProducer.setCustomPartitioner(new LogPartitioner<String>() {
            @Override
            public String getHashKey(String element) {
                String hash = "";
                try {
                    MessageDigest md = MessageDigest.getInstance("MD5");
                    md.update(element.getBytes());
                    hash = new BigInteger(1, md.digest()).toString(16);
                } catch (NoSuchAlgorithmException ignore) {
                }
                StringBuilder builder = new StringBuilder();
                while (builder.length() < 32 - hash.length()) {
                    builder.append("0");
                }
                builder.append(hash);
                return builder.toString();
            }
        });
        stream.addSink(logProducer);
        env.execute("Flink producer");
    }

    public static class EventsGenerator implements SourceFunction<String> {
        private volatile boolean running = true;

        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            long seq = 0;
            while (running) {
                Thread.sleep(10);
                ctx.collect((seq++) + "-" + RandomStringUtils.randomAlphabetic(12));
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}

任务监控

Metric监控

对于Flink SQL任务，Blink控制台提供了非常完备的监控报表，可以通过这些报表观察任务运行的状态，如延迟，内存状态，Failover等。

Metric解释可以参考Blink作业运维相关文档：

https://help.aliyun.com/document_detail/62482.html

注意：SLS Datastream SDK暂时还没有实现Metric上报。

消费进度与监控报警

在SLS查看消费组对应的消费进度。

SLS对于每个消费组会定时输出延迟日志，根据这个延迟日志结合SLS的告警，可以用于监控消费延迟。参考如何开通消费组日志：

https://help.aliyun.com/document_detail/85663.html

任务日志采集

Blink支持把作业日志存储到SLS中，在作业编辑页面右侧的作业参数页面，配置Log4j appender：

log4j.logger.org.apache.hadoop=OFF
log4j.appender.loghub = com.alibaba.blink.log.loghub.BlinkLogHubAppender
log4j.appender.loghub.Threshold = ERROR
log4j.appender.loghub.projectName = <your SLS Project>
log4j.appender.loghub.logstore = <your SLS Logstore>
log4j.appender.loghub.endpoint = <your SLS Endpoint>
log4j.appender.loghub.accessKeyId = <your Access Key ID>
log4j.appender.loghub.accessKey = <your Access Key Secret>

常见问题

部分task没有读到数据

从SLS消费数据的本质就是从每个shard的指定位置开始消费，直到最新位置。然后单个shard不支持并发消费，也就是说一个shard最多被一个task消费到。而不管是SQL还是Data Stream，Shard分配的方式都是对shard本身的信息做hash，然后对task总数取模来分配的。假设某个shard hashcode是x，task 个数为y，当前task的id为z，那么仅当 x%y=z 时，当前task会消费这个shard。因此可能存在某个task没有分配到任何shard的情况。

消费太慢导致数据堆积

数据堆积的根本原因是写入速度超过了消费速度，而如何提高消费速度取决于具体的场景。常见的原因有：

下游处理节点慢导致source节点反压。这种情况可以通过观察Blink的反压状态是否是HIGH来确认，通过优化下游任务节点速度来解决。
Shard个数太少。当shard数量少于task个数时，无疑会造成部分task空跑的现象，此时增加task个数已经对总体的并发没有任何影响，此时可以尝试在SLS侧分裂shard个数解决。

消费组在多个作业之间没有产生分配shard的效果

Blink消费SLS的数据并没有使用消费组来实现均衡消费的效果，提供的消费组名称仅仅是用于在SLS服务端保存消费位点。如果需要类似Kafka 消费组的功能，应使用SLS的Consumer Library。参考文档

https://help.aliyun.com/document_detail/28998.html

Flink Datastream任务没有同步checkpoint到SLS服务端

检查Blink任务作业参数中的 blink.checkpoint.interval.ms 是否设置过大。

基于SLS+Blink的实时计算最佳实践

日志服务简介

阿里云实时计算（Blink）

Flink SQL

创建SLS源表

Checkpoint

创建SLS结果表

Flink Datastream

Flink SQL 的不足

Maven 依赖

消费SLS示例

处理结果写入SLS示例

任务监控

Metric监控

消费进度与监控报警

任务日志采集

常见问题

部分task没有读到数据

消费太慢导致数据堆积

消费组在多个作业之间没有产生分配shard的效果

Flink Datastream任务没有同步checkpoint到SLS服务端

更多资料

继续阅读

Eclipse搭建Web Service服务

web service 的简单实现

Web Service 应用实例

Web Service开发实战

更改LYNC SIP地址

SQL优化SQL语句优化的目的

JAVA高效编程指南

终端环境之tmux

lvm创建、扩容

关于SQL语言

HBuilder开发App Step1——环境搭建，HelloMUI 以及真机调试

SQL语言基础：常用的数据查询语句

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

neo4j之cypher使用文档

【python】【数据处理】画多维数据分布图

sqlServer根据经纬查距离