1.3.2 Spark 基于Structured Streaming的实现

Spark发送数据到Kafka，及最后的执行分析计划，与Flink无区别，不再展开。下面简述差异点。

1. 编写 Spark 任务分析代码

（1）构建 SparkSession
如果需要使用 Spark 的Structured Streaming组件，首先需要创建 SparkSession 实例，代码如下所示：

val sparkConf = new SparkConf()
  .setAppName("StreamingAnalysis")
  .set("spark.local.dir", "F:\\temp")
  .set("spark.default.parallelism", "3")
  .set("spark.sql.shuffle.partitions", "3")
  .set("spark.executor.instances", "3")

val spark = SparkSession
  .builder
  .config(sparkConf)
  .getOrCreate()

（2）从 Kafka 读取答题数据
接下来，从 Kafka 中实时读取答题数，并生成 streaming-DataSet 实例，代码如下所示：

val inputDataFrame1 = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "linux01:9092,linux02:9092,linux03:9092")
  .option("subscribe", "test_topic_learning_1")
  .load()

（3）进行 JSON 解析
从 Kafka 读取到数据后，进行 JSON 解析，并封装到 Answer 实例中，代码如下所示：

val keyValueDataset1 = inputDataFrame1.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)").as[(String, String)]

val answerDS = keyValueDataset1.map(t => {
  val gson = new Gson()
  val answer = gson.fromJson(t._2, classOf[Answer])
  answer
})

其中 Answer 为 Scala 样例类，代码结构如下所示：

case class Answer(student_id: String,
                  textbook_id: String,
                  grade_id: String,
                  subject_id: String,
                  chapter_id: String,
                  question_id: String,
                  score: Int,
                  answer_time: String,
                  ts: Timestamp) extends Serializable

（4）创建临时视图
创建临时视图代码如下所示：

answerDS.createTempView("t_answer")

（5）进行任务分析
仅以需求1（统计题目被作答频次）为例，编写代码如下所示：

实时：统计题目被作答频次

//实时：统计题目被作答频次
val result1 = spark.sql(
  """SELECT
    |  question_id, COUNT(1) AS frequency
    |FROM
    |  t_answer
    |GROUP BY
    |  question_id
  """.stripMargin).toJSON

（6）实时输出分析结果
仅以需求1为例，输出到Kafka 的代码如下所示：

result1.writeStream
 .outputMode("update")
 .trigger(Trigger.ProcessingTime(0))
 .format("kafka")
 .option("kafka.bootstrap.servers", "linux01:9092,linux02:9092,linux03:9092")
 .option("topic", "test_topic_learning_2")
 .option("checkpointLocation", "./checkpoint_chapter11_1")
 .start()

1.3.3 使用 UFlink SQL 加速开发

通过上文可以发现，无论基于Flink还是Spark通过编写代码实现数据分析任务时，都需要编写大量的代码，并且在生产集群上运行时，需要打包程序，然后提交打包后生成的 Jar 文件到集群上运行。

为了简化开发者的工作量，不少开发者开始致力于 SQL 模块的封装，希望能够实现只写 SQL 语句，就完成类似上述的需求。UFlink SQL 即是 UCloud 为简化计算模型、降低用户使用实时计算UFlink产品门槛而推出的一套符合 SQL 语义的开发套件。通过 UFlink SQL 模块可以快速完成这一工作，实践如下。

1. 创建 UKafka 集群

在UCloud控制台UKafka创建页，选择配置并设置相关阈值，创建UKafka集群。

提示：此处暂且忽略在 Kafka 集群中创建 Topic 的操作。

2. 创建 UFlink 集群

在UCloud控制台UFlink创建页，选择配置和运行模式，创建一个 Flink 集群。

完成创建

3. 编写 SQL 语句

完成之后，只需要在工作空间中创建如下形式的 SQL 语句，即可完成上述3个需求分析任务。
（1）创建数据源表
创建数据源表，本质上就是为 Flink 当前上下文环境执行 addSource 操作，SQL 语句如下：

CREATE TABLE t_answer(
    student_id VARCHAR,
    textbook_id VARCHAR,
    grade_id VARCHAR,
    subject_id VARCHAR,
    chapter_id VARCHAR,
    question_id VARCHAR,
    score INT,
    answer_time VARCHAR,
    ts TIMESTAMP
 )WITH(
    type ='kafka11',
    bootstrapServers ='ukafka-mqacnjxk-kafka001:9092,ukafka-mqacnjxk-kafka002:9092,ukafka-mqacnjxk-kafka003:9092',
    zookeeperQuorum ='ukafka-mqacnjxk-kafka001:2181/ukafka',
    topic ='test_topic_learning_1',
    groupId = 'group_consumer_learning_test01',
    parallelism ='3'
 );

（2）创建结果表
创建结果表，本质上就是为 Flink 当前上下文环境执行 addSink 操作，SQL 语句如下：

CREATE TABLE t_result1(
    question_id VARCHAR,
    frequency INT
)WITH(
    type ='kafka11',
    bootstrapServers ='ukafka-mqacnjxk-kafka001:9092,ukafka-mqacnjxk-kafka002:9092,ukafka-mqacnjxk-kafka003:9092',
    zookeeperQuorum ='ukafka-mqacnjxk-kafka001:2181/ukafka',
    topic ='test_topic_learning_2',
    parallelism ='3'
);

CREATE TABLE t_result2(
    grade_id VARCHAR,
    frequency INT
)WITH(
    type ='kafka11',
    bootstrapServers ='ukafka-mqacnjxk-kafka001:9092,ukafka-mqacnjxk-kafka002:9092,ukafka-mqacnjxk-kafka003:9092',
    zookeeperQuorum ='ukafka-mqacnjxk-kafka001:2181/ukafka',
    topic ='test_topic_learning_3',
    parallelism ='3'
);

CREATE TABLE t_result3(
    subject_id VARCHAR,
    question_id VARCHAR,
    frequency INT
)WITH(
    type ='kafka11',
    bootstrapServers ='ukafka-mqacnjxk-kafka001:9092,ukafka-mqacnjxk-kafka002:9092,ukafka-mqacnjxk-kafka003:9092',
    zookeeperQuorum ='ukafka-mqacnjxk-kafka001:2181/ukafka',
    topic ='test_topic_learning_4',
    parallelism ='3'
);

（3）执行查询计划
最后，执行查询计划，并向结果表中插入查询结果，SQL 语句形式如下：

INSERT INTO
    t_result1
  SELECT
      question_id, COUNT(1) AS frequency
    FROM
      t_answer
    GROUP BY
      question_id;

INSERT INTO
    t_result2
  SELECT
      grade_id, COUNT(1) AS frequency
    FROM
      t_answer
    GROUP BY
      grade_id;

INSERT INTO
    t_result3
  SELECT
      subject_id, question_id, COUNT(1) AS frequency
    FROM
      t_answer
    GROUP BY
      subject_id, question_id;

SQL 语句编写完毕后，将其直接粘贴到 UFlink 前端页面对话框中，并提交任务，即可快速完成上述 3 个需求。如下图所示：

1.3.4. UFlink SQL 支持多流 JOIN

Flink、Spark 目前都支持多流 JOIN，即stream-stream join，并且也都支持Watermark处理延迟数据，以上特性均可以在 SQL 中体现，得益于此，UFlink SQL 也同样支持纯 SQL 环境下进行 JOIN 操作、维表JOIN操作、自定义函数操作、JSON数组解析、嵌套JSON解析等。更多细节欢迎大家参考 UFlink SQL 相关案例展示https://docs.ucloud.cn/analysis/uflink/dev/sql

1.4 总结

UFlink 基于 Apache Flink 构建，除100%兼容开源外，也在不断推出 UFlink SQL 等模块，从而提高开发效率，降低使用门槛，在性能、可靠性、易用性上为用户创造价值。今年8月新推出的 Flink 1.9.0，大规模变更了 Flink 架构，能够更好地处理批、流任务，同时引入全新的 SQL 类型系统和更强大的 SQL 式任务编程。UFlink 预计将于10月底支持 Flink 1.9.0，敬请期待。

本文转载自公众号UCloud技术（ID：ucloud_tech）。

原文链接：

https://mp.weixin.qq.com/s/JFcANUK_Vfa7ZMXnn7sruQ

创作场景

实时计算框架 Flink 在教育行业的应用实践（下）