零距离接触Flink:全面解读流计算框架入门与实操指南

# 安装JDK
yum install -y java-1.8.0-openjdk-devel

# 安装Maven
yum install -y maven

wget https://dlcdn.apache.org/flink/flink-1.14.5/flink-1.14.5-bin-scala_2.12.tgz 
tar -xvf flink-1.14.5-bin-scala_2.12.tgz

# 启动JobManager
./bin/start-cluster.sh

# 提交并运行WordCount程序
./bin/flink run examples/streaming/WordCount.jar

// 定义文本源DataStream
DataStream<String> text = env.socketTextStream("localhost", 9999); 

//将每行内容切分转换成单词列表
DataStream<String> words = text
  .flatMap(new FlatMapFunction<String, String>() {
     public void flatMap(String value, Collector<String> out) {
       String[] split = value.toLowerCase().split("\\W+");
       // ...
    }
  });
  
//按单词进行计数统计        
DataStream<Tuple2<String, Long>> counts = words
  .keyBy(value -> value)
  .sum(1);
  
//输出结果
counts.print();

mvn clean package

bin/flink run target/wordcount-1.0-SNAPSHOT.jar

nc localhost 9999
hello world bye
hello again

// 导入Flink相关包
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class WordCount {

  public static void main(String[] args) throws Exception {

    // 创建执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // 从文件读取文本行数据源
    DataStream<String> text = env.addSource(new MySourceFunction());

    // 将每行内容切分成单词
    DataStream<String> words = text.flatMap(new FlatMapFunction<String, String>() {
      public void flatMap(String value, Collector<String> out) {
        String[] splits = value.split("\\s+");
        for (String word : splits) {
          out.collect(word);
        }
      }
    });

    // 按单词进行分组计数
    DataStream<Tuple2<String, Long>> result = words.keyBy(e -> e)
      .timeWindow(Time.seconds(5)) 
      .sum(1);

    // 打印最终结果
    result.print();

    // 执行任务
    env.execute("WordCount");

  }

  // 自定义文本数据源
  public static class MySourceFunction implements SourceFunction<String> {

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
      // 从文件或集合读取文本 
      // ...
      ctx.collect("hello world"); 
    }

    @Override
    public void cancel() {

    }
  }

}

yarn.distributed.enabled: true

mvn package -Pyarn 

./bin/flink run -m yarn-cluster -yn 1 -ys 1 /path/to/job.jar

TableEnvironment tableEnv = TableEnvironment.create(env);

tableEnv.connect(new FlinkKafkaConsumer<>(...)
  .property(...));

CREATE TABLE inputTable (
  id STRING, 
  timestamp TIMESTAMP,
  ...)
WITH (...); 

SELECT 
  id, 
  COUNT(*) 
FROM 
  inputTable
GROUP BY 
  TUMBLE(timestamp, INTERVAL '5' MINUTE)

SELECT 
  SUM(amount) 
FROM 
  inputTable
GROUP BY 
  HOP(timestamp, INTERVAL '1' HOUR, INTERVAL '30' MINUTE)

tableEnv.toRetractStream[Row]...

// 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
TableEnvironment tableEnv = TableEnvironment.create(env); 

// 从Kafka读取文本行数据
tableEnv.connect(new FlinkKafkaConsumer<>(...)
  .topic("kafka_topic"))
  .withFormat(new SimpleStringSchema())
  .createTemporaryTable("lines");

// 分词表
tableEnv.executeSql(
  "CREATE TABLE words WITH ('connector' = 'upsert', 'url' = '...") AS " +
  "SELECT " +
  "  ROW_NUMBER() OVER() AS id, " +
  "  word " + 
  "FROM lines, LATERAL(FLATTEN(SPLIT(lines, ' ')))";

// 窗口聚合表   
tableEnv.executeSql(
  "CREATE TABLE word_counts WITH ('connector' = 'upsert', 'url' = '...'") AS " +
  "SELECT " +
  "  word, " +
  "  COUNT(*) AS count " +
  "FROM words " +
  "GROUP BY TUMBLE(rowtime, INTERVAL '5' SECOND), word");

// 输出结果
tableEnv.executeSql("INSERT INTO sink SELECT * FROM word_counts");

// 执行程序
env.execute();