Simple Log Service:Consume data using Flink

Flink Log Consumer

The Flink Log Consumer subscribes to a Logstore and provides exactly-once semantics. It automatically detects shard changes, which eliminates manual shard management.

Each subtask consumes data from a subset of shards. If shards are split or merged, the subtask's shard assignment updates automatically.

1. Configure startup parameters.

   The following example uses java.util.Properties for configuration. All parameters are defined in the ConfigConstants class.

   Properties configProps = new Properties();
   // The endpoint of Simple Log Service.
   configProps.put(ConfigConstants.LOG_ENDPOINT, "cn-hangzhou.log.aliyuncs.com");
   // In this example, the AccessKey ID and AccessKey secret are obtained from environment variables.
   String accessKeyId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
   String accessKeySecret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
   configProps.put(ConfigConstants.LOG_ACCESSKEYID,accessKeyId);
   configProps.put(ConfigConstants.LOG_ACCESSKEY,accessKeySecret);
   // The Simple Log Service project.
   String project = "your-project";
   // The Simple Log Service Logstore.
   String logstore = "your-logstore";
   // The position from which to start consuming logs.
   configProps.put(ConfigConstants.LOG_CONSUMER_BEGIN_POSITION, Consts.LOG_END_CURSOR);
   // The method to deserialize messages from Simple Log Service.
   FastLogGroupDeserializer deserializer = new FastLogGroupDeserializer();
   final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   DataStream<FastLogGroupList> dataStream = env.addSource(
           new FlinkLogConsumer<FastLogGroupList>(project, logstore, deserializer, configProps)
   );
   dataStream.addSink(new SinkFunction<FastLogGroupList>() {
       @Override
       public void invoke(FastLogGroupList logGroupList, Context context) throws Exception {
           for (FastLogGroup logGroup : logGroupList.getLogGroups()) {
               int logsCount = logGroup.getLogsCount();
               String topic = logGroup.getTopic();
               String source = logGroup.getSource();
               for (int i = 0; i < logsCount; ++i) {
                   FastLog row = logGroup.getLogs(i);
                   for (int j = 0; j < row.getContentsCount(); ++j) {
                       FastLogContent column = row.getContents(j);
                       // Process logs.
                       System.out.println(column.getKey());
                       System.out.println(column.getValue());
                   }
               }
           }
       }
   });
   // Or, use RawLogGroupListDeserializer.
   RawLogGroupListDeserializer rawLogGroupListDeserializer = new RawLogGroupListDeserializer();
   final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   DataStream<RawLogGroupList> rawLogGroupListDataStream = env.addSource(
           new FlinkLogConsumer<RawLogGroupList>(project, logstore, rawLogGroupListDeserializer, configProps)
   );
   rawLogGroupListDataStream.addSink(new SinkFunction<RawLogGroupList>() {
       @Override
       public void invoke(RawLogGroupList logGroupList, Context context) throws Exception {
           for (RawLogGroup logGroup : logGroupList.getRawLogGroups()) {
               String topic = logGroup.getTopic();
               String source = logGroup.getSource();
               for (RawLog row : logGroup.getLogs()) {
                   // Process logs.
               }
           }
       }
   });

   Note

   The number of subtasks is independent of the shard count. If there are more shards than subtasks, each subtask consumes data from a unique set of shards. If there are fewer shards, some subtasks remain idle until new shards are created.

2. Set the start position for consumption.

   Set ConfigConstants.LOG_CONSUMER_BEGIN_POSITION to one of the following values:

   • Consts.LOG_BEGIN_CURSOR: Starts consumption from the beginning of the shard, which is the oldest data in the shard.

   • Consts.LOG_END_CURSOR: Starts consumption from the end of the shard, which is the latest data in the shard.

   • Consts.LOG_FROM_CHECKPOINT: Starts consumption from a checkpoint saved in a specific consumer group. Use ConfigConstants.LOG_CONSUMERGROUP to specify the consumer group.

   • UnixTimestamp: A string that represents a UNIX timestamp in seconds. Consumption starts from the data logged after this timestamp.

   Example:

   configProps.put(ConfigConstants.LOG_CONSUMER_BEGIN_POSITION, Consts.LOG_BEGIN_CURSOR);
   configProps.put(ConfigConstants.LOG_CONSUMER_BEGIN_POSITION, Consts.LOG_END_CURSOR);
   configProps.put(ConfigConstants.LOG_CONSUMER_BEGIN_POSITION, "1512439000");
   configProps.put(ConfigConstants.LOG_CONSUMER_BEGIN_POSITION, Consts.LOG_FROM_CHECKPOINT);

   Note

   If Flink recovers from its own StateBackend, these settings are ignored and consumption resumes from the StateBackend checkpoint.

3. Optional: Set up consumption progress monitoring.

   The Flink Log Consumer supports consumption progress monitoring to retrieve the real-time consumption position of each shard. Step 2: View the status of a consumer group.

   Example:

   configProps.put(ConfigConstants.LOG_CONSUMERGROUP, "your consumer group name");

   Note

   If configured, the Flink Log Consumer creates a consumer group. If the consumer group already exists, no action is taken. Snapshots are automatically synced to the consumer group, and you can view consumption progress in the Simple Log Service console.

4. Configure disaster recovery and exactly-once semantics.

   When Flink checkpointing is enabled, the consumer periodically saves consumption progress. If a task fails, Flink resumes from the latest checkpoint.

   The checkpoint interval determines the maximum data re-consumed on failure:

   final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   // Enable Flink exactly-once semantics.
   env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
   // Save a checkpoint every 5s.
   env.enableCheckpointing(5000);

   Checkpoints in the Flink documentation.

Flink Log Producer

1. Initialize the Flink Log Producer.

   1. Initialize the Properties configuration parameters.

      The initialization is similar to the consumer. The following parameters are available (defaults are used if not specified):

      // The number of I/O threads used to send data. The default value is the number of CPU cores.
      ConfigConstants.IO_THREAD_NUM
      // The maximum time that logs can be cached before being sent. The default value is 2,000 milliseconds.
      ConfigConstants.FLUSH_INTERVAL_MS
      // The total amount of memory that a task can use. The default value is 100 MB.
      ConfigConstants.TOTAL_SIZE_IN_BYTES
      // The maximum blocking time for sending logs when the memory limit is reached. The unit is milliseconds. The default value is 60s.
      ConfigConstants.MAX_BLOCK_TIME_MS
      // The maximum number of retries. The default value is 10.
      ConfigConstants.MAX_RETRIES

   2. Override LogSerializationSchema and define a method to serialize data into a RawLogGroup.

      A RawLogGroup is a collection of logs. Field details are in the Log reference.

      To write data to a specific shard, use LogPartitioner to generate a hash key. If not configured, data is written to random shards.

      For example:

      FlinkLogProducer<String> logProducer = new FlinkLogProducer<String>(new SimpleLogSerializer(), configProps);
      logProducer.setCustomPartitioner(new LogPartitioner<String>() {
            // Generate a 32-bit hash value.
            public String getHashKey(String element) {
                try {
                    MessageDigest md = MessageDigest.getInstance("MD5");
                    md.update(element.getBytes());
                    String hash = new BigInteger(1, md.digest()).toString(16);
                    while(hash.length() < 32) hash = "0" + hash;
                    return hash;
                } catch (NoSuchAlgorithmException e) {
                }
                return  "0000000000000000000000000000000000000000000000000000000000000000";
            }
        });

2. Write simulated data to Simple Log Service:

   // Serialize data into the Simple Log Service data format.
   class SimpleLogSerializer implements LogSerializationSchema<String> {
       public RawLogGroup serialize(String element) {
           RawLogGroup rlg = new RawLogGroup();
           RawLog rl = new RawLog();
           rl.setTime((int)(System.currentTimeMillis() / 1000));
           rl.addContent("message", element);
           rlg.addLog(rl);
           return rlg;
       }
   }
   public class ProducerSample {
       public static String sEndpoint = "cn-hangzhou.log.aliyuncs.com";
       // In this example, the AccessKey ID and AccessKey secret are obtained from environment variables.
       public static String sAccessKeyId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
       public static String sAccessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
       public static String sProject = "ali-cn-hangzhou-sls-admin";
       public static String sLogstore = "test-flink-producer";
       private static final Logger LOG = LoggerFactory.getLogger(ConsumerSample.class);
       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> simpleStringStream = env.addSource(new EventsGenerator());
           Properties configProps = new Properties();
           // The endpoint of Simple Log Service.
           configProps.put(ConfigConstants.LOG_ENDPOINT, sEndpoint);
           // The user's AccessKey.
           configProps.put(ConfigConstants.LOG_ACCESSKEYID, sAccessKeyId);
           configProps.put(ConfigConstants.LOG_ACCESSKEY, sAccessKey);
           // The Simple Log Service project to which logs are written.
           configProps.put(ConfigConstants.LOG_PROJECT, sProject);
           // The Simple Log Service Logstore to which logs are written.
           configProps.put(ConfigConstants.LOG_LOGSTORE, sLogstore);
           FlinkLogProducer<String> logProducer = new FlinkLogProducer<String>(new SimpleLogSerializer(), configProps);
           simpleStringStream.addSink(logProducer);
           env.execute("flink log producer");
       }
       // Simulate log generation.
       public static class EventsGenerator implements SourceFunction<String> {
           private 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;
           }
       }
   }

Consumption example

This example reads data as FastLogGroupList, converts entries to JSON strings with flatMap, and writes the output to a text file.

package com.aliyun.openservices.log.flink.sample;

import com.alibaba.fastjson.JSONObject;
import com.aliyun.openservices.log.common.FastLog;
import com.aliyun.openservices.log.common.FastLogGroup;
import com.aliyun.openservices.log.flink.ConfigConstants;
import com.aliyun.openservices.log.flink.FlinkLogConsumer;
import com.aliyun.openservices.log.flink.data.FastLogGroupDeserializer;
import com.aliyun.openservices.log.flink.data.FastLogGroupList;
import com.aliyun.openservices.log.flink.model.CheckpointMode;
import com.aliyun.openservices.log.flink.util.Consts;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Properties;

public class FlinkConsumerSample {
    private static final String SLS_ENDPOINT = "your-endpoint";
    // In this example, the AccessKey ID and AccessKey secret are obtained from environment variables.
    private static final String ACCESS_KEY_ID = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
    private static final String ACCESS_KEY_SECRET = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
    private static final String SLS_PROJECT = "your-project";
    private static final String SLS_LOGSTORE = "your-logstore";

    public static void main(String[] args) throws Exception {
        final ParameterTool params = ParameterTool.fromArgs(args);

        Configuration conf = new Configuration();
        // Checkpoint dir like "file:///tmp/flink"
        conf.setString(CheckpointingOptions.CHECKPOINTS_DIRECTORY, "your-checkpoint-dir");
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1, conf);
        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("file:///tmp/flinkstate"));
        Properties configProps = new Properties();
        configProps.put(ConfigConstants.LOG_ENDPOINT, SLS_ENDPOINT);
        configProps.put(ConfigConstants.LOG_ACCESSKEYID, 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, "your-consumer-group");
        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);
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("topic", logGroup.getTopic());
                    jsonObject.put("source", logGroup.getSource());
                    for (int j = 0; j < log.getContentsCount(); j++) {
                        jsonObject.put(log.getContents(j).getKey(), log.getContents(j).getValue());
                    }
                    out.collect(jsonObject.toJSONString());
                }
            }
        }).returns(String.class);

        stream.writeAsText("log-" + System.nanoTime());
        env.execute("Flink consumer");
    }
}