大数据框架调研-流处理-Spark与Flink

布满荆棘的人生 2022-09-14 01:46 213阅读 0赞

## 实时处理（流处理） ##

### 结论 ###

Spark和Flink的数据源最好都是Kafka等消息队列，这样才能更好的保证Exactly-Once（精准一次）；

作为流处理框架，Flink是当前最优秀的实时处理框架，并处于飞速发展的状态中；

Spark社区活跃度高，生态圈庞大，Spark-Streaming技术成熟稳定，且Spark是批处理框架中使用最为广泛的框架，如果需要批处理的情况下，批处理和流处理都是用Spark，可以大大减少框架的学习成本，并且不需要不同的框架，因此在没有特殊情况时，推荐使用Spark-Streaming作为流处理的框架。

Spark框架可以作用于几乎所有大数据的任务。

如果需要完美的话，以下一些情况推荐使用Flink：

1、对数据要求是毫秒级的实时处理场景；

2、必须按照事件时间语义的流处理场景；

### 数据源（实时数据） ###

**Spark**：自定义数据输入，最好是Kafka等消息队列；

**Flink**：自定义数据输入，最好是Kafka等消息队列；

### 计算引擎 ###

**Spark**:

Spark Streaming，通过**Java或者scala**编写计算代码，属于微批次处理，运行的时候需要指定批处理的时间，每次运行 job 时处理一个批次的数据，实时性还不够；

实际上每个job都可以理解为一个Spark-core任务，每次处理一个job都是先缓存数据，然后进行批处理的；

Spark Streaming 只支持处理时间，Structured streaming（Spark的结构化微批次处理框架，还停留在beta阶段，因此官方声明，仅供用户学习、实验和测试） 支持处理时间和事件时间，同时支持 watermark 机制处理滞后数据；因此Streaming本身没办法处理事件时间语义的数据，且对无法针对乱序数据和滞后进行处理；

SparkStreming通过checkPoint机制与Kafka保证数据输入的Exactly-Once，数据写出的Exactly-Once处理需要自定义；

SparkStreaming只支持基于时间的窗口操作（处理时间或者事件时间）；

SparkStreaming社区活跃度高，技术成熟稳定。

**Flink**：

Flink ，通过**Java或者scala**编写计算代码，是基于**事件**驱动的，事件可以理解为消息。事件驱动的应用程序是一种状态应用程序，它会从一个或者多个流中注入事件，通过触发计算更新状态，或外部动作对注入的事件作出反应；

事件数据会在整个Flink任务中流动处理，一个处理完成之后立马处理下一个；

Flink 支持三种时间机制：事件时间，注入时间，处理时间，同时支持 watermark 机制处理滞后数据；支持对乱序数据和滞后数据进行处理；

Flink通过checkPoint机制与Kafka保证数据输入的Exactly-Once，也可以Kafka作为输出就可以保证输出的Exactly-Once；

Flink支持的窗口操作非常灵活，不仅支持时间窗口，还支持基于数据本身的窗口(另外还支持基于time、count、session，以及data-driven的窗口操作)，可以自由定义想要的窗口操作；

Flink的每个任务都会有对应的webUI查看，并且可以通过webUI对应的api接口，在我们自己的程序中获取到对应任务的一些信息，非常方便对任务的管理。

Flink处于高速发展阶段，社区活跃度不如SparkStreaming。

### 数据输出 ###

**Spark**:Spark-Streaming可以选择输出到文件或者自定义数据输出，通过转换成DataFrame数据结构可以很方便的将数据输出到数据库或是其他结构化文件中

**Flink**：支持输出到文件、Redis、Netty、Kafka等，也可以自定义数据输出。

### 执行效率 ###

**Spark**：基于内存的计算，属于微批次低延迟的模拟流处理；

**Flink**：基于内存的计算，真正的流处理。

### 任务提交 ###

**Spark**：

Spark的提交方法jar包+Shell命令、jar包+Spark-launcher、ssh命令

jar包+Shell命令、jar包+Spark-launcher方法均需要依赖小程序，后者通过launcher能够更好的对Spark任务进行管理；

ssh命令直接提交方法不依赖小程序，但是对于任务的管理更加不方便；

Spark 提供了以Spark-launcher 作为JAVA API编程的方式提交，这种方式不需要使用命令行，能够实现与Spring整合，让应用在tomcat中运行；Spark-launcher值支持将任务运行，日志分开输出，便于问题的回溯，可以自定义监听器，当信息或者状态变更时，进行相关操作，支持暂停、停止、断连、获得AppId、获得任务State等多种功能。

可以通过第三方组件对Spark-Streaming任务进行优雅退出。

**Flink**

Flink的提交方法包括jar包+Shell命令、jar包+Flink-Yarn、ssh命令

ar包+Shell命令、jar包+Spark-launcher方法均需要依赖小程序，后者通过launcher能够更好的对Spark任务进行管理；

ssh命令直接提交方法不依赖小程序，但是对于任务的管理更加不方便；

Flink-Yarn方式可以通过JAVA API远程进行任务的提交，不需要依赖小程序和命令行，能够实现与Spring整合，让应用在tomcat中运行；相比通过命令行方式，能够直接通过api获取到webUI地址以及applicationId，可以更加方便的进行任务的管理。

可以通过yarn方式，或者webUI的api接口来退出Flink任务。

# 实时处理 #

## 测试环境 ##

Kafka作为数据源，数据结果均输出到数据库中。

Hadoop的Yarn作为任务运行的资源调度管理器（spark on yarn 和 flink on yarn）。

## 数据获取与采集 ##

实时数据可以通过以下方式进行采集：

1、可以通过Flume监控日志文件的方式，然后数据既传给HDFS进行保存，又将数据发送给Kafka；

2、通过Kafka消息队列进行数据的获取，Flume作为Kafka的其中一个消费者将数据保存到HDFS保存；

本次实验只涉及Kafka的部分，Kafka的topic为streamTest，然后以控制台输入作为Kafka的数据生产者

## 创建topic
    kafka-topics.sh --zookeeper hadoop113:2181 --create --topic streamTest --replication-factor 2 --partitions 2
    
    ## 控制台生产者
    kafka-console-producer.sh --broker-list localhost:9092 --topic streamTest

## job提交与运行 ##

### Spark-Streaming ###

object WordCountByStream { 
    
        //初始化连接池
        var dataSource: DataSource = init()
    
        def main(args: Array[String]): Unit = { 
    
            // 创建 SparkConf
            val sparkConf: SparkConf = new SparkConf().setAppName("ReceiverWordCount").setMaster("yarn")
    
            //创建 StreamingContext
            val ssc = new StreamingContext(sparkConf, Seconds(3))
    
            // 定义 Kafka 参数
            val kafkaPara: Map[String, Object] = Map[String, Object](
                ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop113:9092,hadoop114:9092,hadoop115:9092",
                ConsumerConfig.GROUP_ID_CONFIG -> "sparkStreamTestGroup",
                ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer",
                ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer",
                ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "latest"
            )
    
            // 读取 Kafka 数据创建 DStream
            val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
                ssc,
                LocationStrategies.PreferConsistent,
                ConsumerStrategies.Subscribe[String, String](Set("streamTest"), kafkaPara))
    
            val wordCountDS: DStream[(String, Long)] = kafkaDStream.flatMap(
                kafkaData => { 
                    val data = kafkaData.value()
                    val datas: Array[String] = data.split(" ")
                    datas.map(str => str -> 1L)
                }
            ).reduceByKey(_ + _)
    
            // 输出
            wordCountDS.foreachRDD(
                rdd => { 
                    rdd.foreachPartition(
                        iter => { 
                            // 此处是放入数据库中，可以在这里设置放到其他地方，如redis等
                            val conn = getConnection
                            val pstat = conn.prepareStatement(
                                """ | insert into spark_streaming_test | (word, cnt) | values | (?, ?) | on duplicate key | update cnt = cnt + ? |""".stripMargin)
                            iter.foreach { 
                                case (word, count) => { 
    
                                    println(s"$word, $count")
    
                                    pstat.setString(1, word)
                                    pstat.setLong(2, count)
                                    pstat.setLong(3, count)
    
                                    pstat.executeUpdate()
                                }
                            }
                            pstat.close()
                            conn.close()
                        }
                    )
                }
            )
    
            // 开启任务
            ssc.start()
    
            // 优雅退出
            stopHandle(ssc)
            // 这个是批量处理的退出，用Ctrl+C来退出
            // ssc.awaitTermination()
        }
    
        def stopHandle(ssc: StreamingContext): Unit = { 
            // 优雅的关闭
            // 计算节点不再接受新的数据，而是将现有的数据处理完毕，然后关闭
            // mysql、redis、zk、hdfs等
            var needStop = false;
            while (true) { 
                // 判断是否需要关闭
                if (needStop) { 
                    if (ssc.getState() == StreamingContextState.ACTIVE) { 
                        ssc.stop(true, true)
                        System.exit(0)
                    }
                }
                Thread.sleep(10000)
                needStop = true
            }
        }
    
        //初始化连接池方法
        def init(): DataSource = { 
    
            val properties = new Properties()
            properties.setProperty("driverClassName", "com.mysql.jdbc.Driver")
            properties.setProperty("url", "jdbc:mysql://10.10.10.38:13306/stream_test?useUnicode=true&characterEncoding=UTF-8")
            properties.setProperty("username", "root")
            properties.setProperty("password", "123456")
            properties.setProperty("maxActive", "50")
            DruidDataSourceFactory.createDataSource(properties)
        }
    
        //获取 MySQL 连接
        def getConnection: Connection = { 
    
            dataSource.getConnection
        }
    }

那么可以和Spark的方法完全一样，将以上代码达成jar包，然后提交运行即可。

/opt/module/spark-yarn/bin/spark-submit --class com.starnet.server.bigdata.spark.stream.wordcount.WordCountByStream  --master yarn  --deploy-mode cluster /home/bd/SPARK/spark-test-1.0.0.jar

或者是通过Spark-launcher的方式进行提交。

在Yarn上运行时，可以提前将所使用的jar包全部上传到HDFS然后进行jar路径的配置，可以使每次任务提交时不用重新上次依赖文件，Flink也是如此。

### Flink ###

public class WordCount { 
    
        public static void main(String[] args) throws Exception { 
    
            // 创建流处理执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            // 创建kafka配置对象
            Properties properties = new Properties();
            properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop113:9092,hadoop114:9092,hadoop115:9092");
            properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "flinkTestGroup");
            properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.StringDeserializer");
            properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.StringDeserializer");
            properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
    
            // 从Kafka读取数据
            DataStreamSource<String> inputDataStream = env.addSource(new FlinkKafkaConsumer010<String>(
                    "streamTest", new SimpleStringSchema(), properties));
    
            // 基于数据流记性转换计算
            DataStream<Tuple2<String, Integer>> resultStream = inputDataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() { 
                @Override
                public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception { 
                String[] fields = s.split(" ");
                for (String  field: fields) { 
                    collector.collect(new Tuple2<String, Integer>(field, 1));
                }
    
                }
            }).keyBy(0).sum(1);
    
            resultStream.addSink(new MyJdbcSink());
    
    
            // 执行任务
            env.execute();
        }
    
        public static class MyJdbcSink extends RichSinkFunction<Tuple2<String, Integer>> { 
    
            Connection connection = null;
            PreparedStatement updateStmt = null;
    
            @Override
            public void open(Configuration parameters) throws Exception { 
    
                connection = DriverManager.getConnection("jdbc:mysql://10.10.10.38:13306/stream_test",
                        "root",
                        "123456");
    
                updateStmt = connection.prepareStatement(
                        "insert into flink_test (word, cnt) " +
                        "values " +
                        "(?, ?) " +
                        "on duplicate key " +
                        "update cnt = cnt + ?");
            }
    
            // 每次更新数据时，调用连接执行sql
            @Override
            public void invoke(Tuple2<String, Integer> value, Context context) throws Exception { 
    
                System.out.println(value.toString());
    
                updateStmt.setString(1, value.f0);
                updateStmt.setLong(2, value.f1);
                updateStmt.setLong(3, value.f1);
    
                updateStmt.execute();
    
            }
    
            @Override
            public void close() throws Exception { 
    
                connection.close();
                updateStmt.close();
            }
        }
    }

#### Flink Per Job ####

那么可以将以上代码打成jar包，然后提交运行的方式进行处理。

bin/flink run –m yarn-cluster -c com.starnet.server.bigdata.flink.WordCount /home/bd/FLINK/FlinkTest-1.0-SNAPSHOT-jar-with-dependencies.jar

通过jar提交成功之后，控制台输出如下：

2021-09-26 09:51:58,784 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - YARN application has been deployed successfully.
    2021-09-26 09:51:58,785 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - The Flink YARN session cluster has been started in detached mode. In order to stop Flink gracefully, use the following command:
    $ echo "stop" | ./bin/yarn-session.sh -id application_1631184398602_0052
    If this should not be possible, then you can also kill Flink via YARN's web interface or via:
    $ yarn application -kill application_1631184398602_0052
    Note that killing Flink might not clean up all job artifacts and temporary files.
    2021-09-26 09:51:58,785 INFO  org.apache.flink.yarn.YarnClusterDescriptor                   - Found Web Interface hadoop114:39315 of application 'application_1631184398602_0052'.
    Job has been submitted with JobID 83ea15f02379ee196c730d6919f93243

其中可以通过以下两个命令进行任务的关闭

echo "stop" | /.../flink/.../bin/yarn-session.sh -id application_id
    
    /.../hadoop/.../yarn application -kill application_id

并且其中有`hadoop114:39315`为提交flink任务之后，在yarn上开启的flink集群的web地址，可以在web上进行任务的关闭或者任务信息的获取等，也可以通过Flink的客户端api进行信息的获取和任务的停止[https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/ops/rest\_api/][https_ci.apache.org_projects_flink_flink-docs-release-1.13_docs_ops_rest_api]

以上信息均是在提交jar进程完成之后去进行信息的过滤和提取，提取出我们需要的信息，然后进行任务的退出。

目前Flink-on-yarn的per job模式只找到了通过jar包提交的方法。

#### Flink Application ####

对于per job模式，jar包的解析、生成JobGraph是在客户端上执行的，然后将生成的jobgraph提交到集群。如果任务特别多的话，那么这些生成JobGraph、提交到集群的操作都会在实时平台所在的机器上执行，那么将会给服务器造成很大的压力。

此外这种模式提交任务的时候会把本地flink的所有jar包先上传到hdfs上相应 的临时目录，这个也会带来大量的网络的开销，所以如果任务特别多的情况下，平台的吞吐量将会直线下降。

所以针对flink per job模式的一些问题，flink 引入了一个新的部署模式–Application模式。 目前 Application 模式支持 Yarn 和 K8s 的部署方式，Yarn Application 模式会在客户端将运行任务需要的依赖都上传到 Flink Master，然后在 Master 端进行任务的提交。

此外，还支持远程的用户jar包来提交任务，比如可以将jar放到hdfs上，进一步减少上传jar所需的时间，从而减少部署作业的时间。

提交方法如下：

**jar包提交**

那么可以将以上代码打成jar包，然后提交运行的方式进行处理。

bin/flink run -yd -m yarn-cluster -c com.starnet.server.bigdata.flink.WordCount /home/bd/FLINK/FlinkTest-1.0-SNAPSHOT-jar-with-dependencies.jar
    
    /opt/module/flink-1.11.4/bin/flink run -yd -m yarn-cluster -c com.starnet.server.bigdata.flink.WordCount -yD yarn.provided.lib.dirs="hdfs://hadoop113:8020/jar/flink11/libs" /home/bd/FLINK/Flink1.11-1.0-SNAPSHOT-jar-with-dependencies.jar

通过jar提交成功之后，控制台输出如下：

2021-09-26 18:00:54,129 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - YARN application has been deployed successfully.
    2021-09-26 18:00:54,529 INFO  org.apache.flink.yarn.YarnClusterDescriptor                  [] - Found Web Interface hadoop115:35553 of application 'application_1631184398602_0055'.

优雅退出和web访问均与以上相同。

**java-api方式提交**

提交相关代码如下：

public void crateStreamTaskByFlinkClient() { 
    
        //flink的本地配置目录，为了得到flink的配置
        // 如果出现org.apache.flink.streaming.runtime.tasks.StreamTaskException: Cannot instantiate user function.错误
        // 则在flink-config.yaml加入
        // classloader.resolve-order: parent-first
        String configurationDirectory = "/opt/module/flink-1.11.4/conf";
        // String configurationDirectory = "/home/lxj/workspace/Olt-Test/bigdata/bigdataserver/src/main/resources/flink/conf";
    
        //存放flink集群相关的jar包目录
        String flinkLibs = "hdfs://hadoop113:8020/jar/flink11/libs";
        //用户jar
        String userJarPath = "hdfs://hadoop113:8020/jar/userTask/Flink1.11-1.0-SNAPSHOT-jar-with-dependencies.jar";
        String flinkDistJar = "hdfs://hadoop113:8020/jar/flink11/libs/flink-dist_2.12-1.11.4.jar";
    
        YarnClient yarnClient = YarnClient.createYarnClient();
        YarnConfiguration yarnConfiguration = new YarnConfiguration();
        yarnClient.init(yarnConfiguration);
        yarnClient.start();
    
        // 设置日志的，没有的话看不到日志
        YarnClusterInformationRetriever clusterInformationRetriever = YarnClientYarnClusterInformationRetriever
            .create(yarnClient);
    
        //获取flink的配置
        Configuration flinkConfiguration = GlobalConfiguration.loadConfiguration(
            configurationDirectory);
    
        flinkConfiguration.set(CheckpointingOptions.INCREMENTAL_CHECKPOINTS, true);
    
        flinkConfiguration.set(
            PipelineOptions.JARS,
            Collections.singletonList(userJarPath));
    
        Path remoteLib = new Path(flinkLibs);
        flinkConfiguration.set(
            YarnConfigOptions.PROVIDED_LIB_DIRS,
            Collections.singletonList(remoteLib.toString()));
    
        flinkConfiguration.set(
            YarnConfigOptions.FLINK_DIST_JAR,
            flinkDistJar);
    
        // 设置为application模式
        flinkConfiguration.set(
            DeploymentOptions.TARGET,
            YarnDeploymentTarget.APPLICATION.getName());
    
        // yarn application name
        flinkConfiguration.set(YarnConfigOptions.APPLICATION_NAME, "flink-application");
    
        YarnLogConfigUtil.setLogConfigFileInConfig(flinkConfiguration, configurationDirectory);
    
        ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder()
            .createClusterSpecification();
    
        // 设置用户jar的参数和主类
        ApplicationConfiguration appConfig = new ApplicationConfiguration(new String[] { "test"}, "com.starnet.server.bigdata.flink.WordCount");
    
        YarnClusterDescriptor yarnClusterDescriptor = new YarnClusterDescriptor(
            flinkConfiguration,
            yarnConfiguration,
            yarnClient,
            clusterInformationRetriever,
            true);
    
        try { 
            ClusterClientProvider<ApplicationId> clusterClientProvider = yarnClusterDescriptor.deployApplicationCluster(
                clusterSpecification,
                appConfig);
    
            ClusterClient<ApplicationId> clusterClient = clusterClientProvider.getClusterClient();
    
            ApplicationId applicationId = clusterClient.getClusterId();
            String webInterfaceURL = clusterClient.getWebInterfaceURL();
            log.error("applicationId is {}", applicationId);
            log.error("webInterfaceURL is {}", webInterfaceURL);
            
            // 退出
            // yarnClusterDescriptor.killCluster(applicationId);
        } catch (Exception e){ 
            log.error(e.getMessage(), e);
        }
    }

此方法可以远程提交Flink任务到yarn上运行，并且可以通过javaApi获取到提交之后的applicationId和web地址，以及任务的退出，整体可以不依赖小程序，非常的方便。

## 结果获取 ##

以上结果最终都是保存如数据库了，因此本次实验的流式处理结果均以查询数据库的方式获取结果。

[https_ci.apache.org_projects_flink_flink-docs-release-1.13_docs_ops_rest_api]: https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/ops/rest_api/