2022-03-20 9647355a5c84dbefa64d859f7aa22817 99+ 3 m 0.4 k

处理无界和有界数据

https://flink.apache.org/zh/flink-architecture.html

任何类型的数据都可以形成一种事件流。信用卡交易、传感器测量、机器日志、网站或移动应用程序上的用户交互记录，所有这些数据都形成一种流。

数据可以被作为无界或者有界流来处理。

无界流 有定义流的开始，但没有定义流的结束。它们会无休止地产生数据。无界流的数据必须持续处理，即数据被摄取后需要立刻处理。我们不能等到所有数据都到达再处理，因为输入是无限的，在任何时候输入都不会完成。处理无界数据通常要求以特定顺序摄取事件，例如事件发生的顺序，以便能够推断结果的完整性。
有界流 有定义流的开始，也有定义流的结束。有界流可以在摄取所有数据后再进行计算。有界流所有数据可以被排序，所以并不需要有序摄取。有界流处理通常被称为批处理

Apache Flink 擅长处理无界和有界数据集精确的时间控制和状态化使得 Flink 的运行时(runtime)能够运行任何处理无界流的应用。有界流则由一些专为固定大小数据集特殊设计的算法和数据结构进行内部处理，产生了出色的性能。

大数据基础组件 flink 原理

处理无界和有界数据

2022-03-20 1c07750d99ce86c104a5c97bef42d48f 99+ 2 m 0.3 k

Flink程序构成部分

⚫ 获取执行环境（ execution environment）
⚫ 读取数据源（ source）
⚫ 定义基于数据的转换操作（ transformations）
⚫ 定义计算结果的输出位置（ sink）

1 source

1 从集合中读取数据

最简单的读取数据的方式，就是在代码中直接创建一个Java集合，然后调用执行环境fromCollection方法进行读取。这相当于将数据临时存储到内存中，形成特殊的数据结构后，作为数据源使用，一般用于测试。

2 从文件读取数据

真正的实际应用中，自然不会直接将数据写在代码中。通常情况下，我们会从存储介质中获取数据，一个比较常见的方式就是读取日志文件。这也是批处理中最常见的读取方式。

1	DataStream<String> stream = env.readTextFile("clicks .csv“);

3 socket

https://www.jianshu.com/p/cb26a0f6c622

socket文本流的读取需要配置两个参数：发送端主机名和端口

文本流数据的发送，可以通过 Linux系统自带的 netcat工具进行模拟。

1	nc -lk 7777

4 kafka

5 自定义 Source

3 sink

1 输出到文件

2 输出到 Kafka

3 输出到 Redis

4 输出到 Elasticsearch

5 输出到 MySQL (JDBC)

6 自定义 Sink输出

大数据基础组件 flink 使用

Flink程序构成部分

2022-03-19 d68cf54b7eb2e68e5ebcdd3d8163ccf9 99+ fast 0.0 k

spark交互工具

https://blog.csdn.net/u010886217/article/details/82916401

spark-shell、spark-sql，thriftserver

大数据基础组件 spark 使用

spark交互工具

2022-03-18 984c49e4fd9ad56487b01a3dca5c09b6 99+ fast 0.0 k

sql,hql区别

https://www.geeksforgeeks.org/difference-between-sql-and-hiveql/

大数据基础组件数据库关系型 sql

sql,hql区别

2022-03-17 47acde4d4f107844153e2d170e54163c 99+ fast 0.0 k

hive数据导入导出

https://www.cnblogs.com/xing901022/p/5801061.html

大数据基础组件 hive hql

hive数据导入导出

2022-03-16 b960c2ccd459bad3ab265e11032cc652 99+ 13 m 2.0 k

spark资源参数调优

https://www.cnblogs.com/gtscool/p/13072051.html

https://blog.csdn.net/l1394049664/article/details/81811642

https://tech.meituan.com/2016/04/29/spark-tuning-basic.html

性能

task

设置分区数

参数说明：该参数用于设置每个stage的默认task数量。这个参数极为重要，如果不设置可能会直接影响你的Spark作业性能。
参数调优建议：Spark作业的默认task数量为500~1000个较为合适。很多同学常犯的一个错误就是不去设置这个参数，那么此时就会导致Spark自己根据底层HDFS的block数量来设置task的数量，默认是一个HDFS block对应一个task。通常来说，Spark默认设置的数量是偏少的（比如就几十个task），如果task数量偏少的话，就会导致你前面设置好的Executor的参数都前功尽弃。试想一下，无论你的Executor进程有多少个，内存和CPU有多大，但是task只有1个或者10个，那么90%的Executor进程可能根本就没有task执行，也就是白白浪费了资源！因此Spark官网建议的设置原则是，设置该参数为num-executors * executor-cores的2~3倍较为合适，比如Executor的总CPU core数量为300个，那么设置1000个task是可以的，此时可以充分地利用Spark集群的资源。

worker

几个执行节点，一个worker 一般对应1个 executor

executor

num-executors

参数说明：该参数用于设置Spark作业总共要用多少个Executor进程来执行。Driver在向YARN集群管理器申请资源时，YARN集群管理器会尽可能按照你的设置来在集群的各个工作节点上，启动相应数量的Executor进程。这个参数非常之重要，如果不设置的话，默认只会给你启动少量的Executor进程，此时你的Spark作业的运行速度是非常慢的。
参数调优建议：每个Spark作业的运行一般设置50~100个左右的Executor进程比较合适，设置太少或太多的Executor进程都不好。设置的太少，无法充分利用集群资源；设置的太多的话，大部分队列可能无法给予充分的资源。

executor-cores

参数说明：该参数用于设置每个Executor进程的CPU core数量。这个参数决定了每个Executor进程并行执行task线程的能力。因为每个CPU core同一时间只能执行一个task线程，因此每个Executor进程的CPU core数量越多，越能够快速地执行完分配给自己的所有task线程。
参数调优建议：Executor的CPU core数量设置为2~4个较为合适。同样得根据不同部门的资源队列来定，可以看看自己的资源队列的最大CPU core限制是多少，再依据设置的Executor数量，来决定每个Executor进程可以分配到几个CPU core。同样建议，如果是跟他人共享这个队列，那么num-executors * executor-cores不要超过队列总CPU core的1/3~1/2左右比较合适，也是避免影响其他同学的作业运行。

内存

driver

driver-memory

参数说明：该参数用于设置Driver进程的内存。
参数调优建议：Driver的内存通常来说不设置，或者设置1G左右应该就够了。唯一需要注意的一点是，如果需要使用collect算子将RDD的数据全部拉取到Driver上进行处理，那么必须确保Driver的内存足够大，否则会出现OOM内存溢出的问题。

executor

Executor的内存主要分为三块：第一块是让task执行我们自己编写的代码时使用，默认是占Executor总内存的20%；第二块是让task通过shuffle过程拉取了上一个stage的task的输出后，进行聚合等操作时使用，默认也是占Executor总内存的20%；第三块是让RDD持久化时使用，默认占Executor总内存的60%。

executor-memory

参数说明：该参数用于设置每个Executor进程的内存。Executor内存的大小，很多时候直接决定了Spark作业的性能，而且跟常见的JVM OOM异常，也有直接的关联。
参数调优建议：每个Executor进程的内存设置4G~8G较为合适。但是这只是一个参考值，具体的设置还是得根据不同部门的资源队列来定。可以看看自己团队的资源队列的最大内存限制是多少，num-executors乘以executor-memory，是不能超过队列的最大内存量的。此外，如果你是跟团队里其他人共享这个资源队列，那么申请的内存量最好不要超过资源队列最大总内存的1/3~1/2，避免你自己的Spark作业占用了队列所有的资源，导致别的同学的作业无法运行。

spark.shuffle.memoryFraction

参数说明：该参数用于设置shuffle过程中一个task拉取到上个stage的task的输出后，进行聚合操作时能够使用的Executor内存的比例，默认是0.2。也就是说，Executor默认只有20%的内存用来进行该操作。shuffle操作在进行聚合时，如果发现使用的内存超出了这个20%的限制，那么多余的数据就会溢写到磁盘文件中去，此时就会极大地降低性能。
参数调优建议：如果Spark作业中的RDD持久化操作较少，shuffle操作较多时，建议降低持久化操作的内存占比，提高shuffle操作的内存占比比例，避免shuffle过程中数据过多时内存不够用，必须溢写到磁盘上，降低了性能。此外，如果发现作业由于频繁的gc导致运行缓慢，意味着task执行用户代码的内存不够用，那么同样建议调低这个参数的值。

spark.storage.memoryFraction

参数说明：该参数用于设置RDD持久化数据在Executor内存中能占的比例，默认是0.6。也就是说，默认Executor 60%的内存，可以用来保存持久化的RDD数据。根据你选择的不同的持久化策略，如果内存不够时，可能数据就不会持久化，或者数据会写入磁盘。
参数调优建议：如果Spark作业中，有较多的RDD持久化操作，该参数的值可以适当提高一些，保证持久化的数据能够容纳在内存中。避免内存不够缓存所有的数据，导致数据只能写入磁盘中，降低了性能。但是如果Spark作业中的shuffle类操作比较多，而持久化操作比较少，那么这个参数的值适当降低一些比较合适。此外，如果发现作业由于频繁的gc导致运行缓慢（通过spark web ui可以观察到作业的gc耗时），意味着task执行用户代码的内存不够用，那么同样建议调低这个参数的值。

资源参数参考示例

以下是一份spark-submit命令的示例，大家可以参考一下，并根据自己的实际情况进行调节：

./bin/spark-submit \
  --master yarn-cluster \
  --num-executors 100 \
  --executor-memory 6G \
  --executor-cores 4 \
  --driver-memory 1G \
  --conf spark.default.parallelism=1000 \
  --conf spark.storage.memoryFraction=0.5 \
  --conf spark.shuffle.memoryFraction=0.3 \

大数据基础组件 spark 优化

spark资源参数调优

2022-03-14 59243e86bd239cdb7140362c55540e3b 99+ 5 m 0.8 k

flink提交任务

https://codeantenna.com/a/Y6wpSYwfRL

1.web ui

https://blog.csdn.net/godelgnis/article/details/106051751

2.命令行

可以指定部署方式

https://nightlies.apache.org/flink/flink-docs-release-1.13/zh/docs/deployment/cli/

https://blog.csdn.net/weixin_42993799/article/details/106566037

1 参数必选 ： 
     -n,--container <arg>   分配多少个yarn容器 (=taskmanager的数量)  
2 参数可选 ： 
     -D <arg>                        动态属性  
     -d,--detached                   独立运行  
     -jm,--jobManagerMemory <arg>    JobManager的内存 [in MB]  
     -nm,--name                      在YARN上为一个自定义的应用设置一个名字  
     -q,--query                      显示yarn中可用的资源 (内存, cpu核数)  
     -qu,--queue <arg>               指定YARN队列.  
     -s,--slots <arg>                每个TaskManager使用的slots数量  
     -tm,--taskManagerMemory <arg>   每个TaskManager的内存 [in MB]  
     -z,--zookeeperNamespace <arg>   针对HA模式在zookeeper上创建NameSpace 
     -id,--applicationId <yarnAppId> YARN集群上的任务id，附着到一个后台运行的yarn session中
 
3 run [OPTIONS] <jar-file> <arguments>  
 
    run操作参数:  
    -c,--class <classname>  如果没有在jar包中指定入口类，则需要在这里通过这个参数指定  
    -m,--jobmanager <host:port>  指定需要连接的jobmanager(主节点)地址，使用这个参数可以指定一个不同于配置文件中的jobmanager  
    -p,--parallelism <parallelism>   指定程序的并行度。可以覆盖配置文件中的默认值。
 
4 启动一个新的yarn-session,它们都有一个y或者yarn的前缀
 
    例如：./bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar 
    
    连接指定host和port的jobmanager：
    ./bin/flink run -m SparkMaster:1234 ./examples/batch/WordCount.jar -input hdfs://hostname:port/hello.txt -output hdfs://hostname:port/result1
 
    启动一个新的yarn-session：
    ./bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar -input hdfs://hostname:port/hello.txt -output hdfs://hostname:port/result1
 
5 注意：命令行的选项也可以使用./bin/flink 工具获得。
 
6 Action "run" compiles and runs a program.
    
      Syntax: run [OPTIONS] <jar-file> <arguments>
      "run" action options:
         -c,--class <classname>               Class with the program entry point
                                              ("main" method or "getPlan()" method.
                                              Only needed if the JAR file does not
                                              specify the class in its manifest.
         -C,--classpath <url>                 Adds a URL to each user code
                                              classloader  on all nodes in the
                                              cluster. The paths must specify a
                                              protocol (e.g. file://) and be
                                              accessible on all nodes (e.g. by means
                                              of a NFS share). You can use this
                                              option multiple times for specifying
                                              more than one URL. The protocol must
                                              be supported by the {@link
                                              java.net.URLClassLoader}.
         -d,--detached                        If present, runs the job in detached
                                              mode
         -n,--allowNonRestoredState           Allow to skip savepoint state that
                                              cannot be restored. You need to allow
                                              this if you removed an operator from
                                              your program that was part of the
                                              program when the savepoint was
                                              triggered.
         -p,--parallelism <parallelism>       The parallelism with which to run the
                                              program. Optional flag to override the
                                              default value specified in the
                                              configuration.
         -q,--sysoutLogging                   If present, suppress logging output to
                                              standard out.
         -s,--fromSavepoint <savepointPath>   Path to a savepoint to restore the job
                                              from (for example
                                              hdfs:///flink/savepoint-1537).
 
7  Options for yarn-cluster mode:
         -d,--detached                        If present, runs the job in detached
                                              mode
         -m,--jobmanager <arg>                Address of the JobManager (master) to
                                              which to connect. Use this flag to
                                              connect to a different JobManager than
                                              the one specified in the
                                              configuration.
         -yD <property=value>                 use value for given property
         -yd,--yarndetached                   If present, runs the job in detached
                                              mode (deprecated; use non-YARN
                                              specific option instead)
         -yh,--yarnhelp                       Help for the Yarn session CLI.
         -yid,--yarnapplicationId <arg>       Attach to running YARN session
         -yj,--yarnjar <arg>                  Path to Flink jar file
         -yjm,--yarnjobManagerMemory <arg>    Memory for JobManager Container with
                                              optional unit (default: MB)
         -yn,--yarncontainer <arg>            Number of YARN container to allocate
                                              (=Number of Task Managers)
         -ynl,--yarnnodeLabel <arg>           Specify YARN node label for the YARN
                                              application
         -ynm,--yarnname <arg>                Set a custom name for the application
                                              on YARN
         -yq,--yarnquery                      Display available YARN resources
                                              (memory, cores)
         -yqu,--yarnqueue <arg>               Specify YARN queue.
         -ys,--yarnslots <arg>                Number of slots per TaskManager
         -yst,--yarnstreaming                 Start Flink in streaming mode
         -yt,--yarnship <arg>                 Ship files in the specified directory
                                              (t for transfer)
         -ytm,--yarntaskManagerMemory <arg>   Memory per TaskManager Container with
                                              optional unit (default: MB)
         -yz,--yarnzookeeperNamespace <arg>   Namespace to create the Zookeeper
                                              sub-paths for high availability mode
         -z,--zookeeperNamespace <arg>        Namespace to create the Zookeeper
                                              sub-paths for high availability mode

大数据基础组件 flink 使用

flink提交任务

2022-03-12 d8cee0e274ae9b92638d35caa51e4aaa 99+ 2 m 0.3 k

flink部署

Flink的部署方式是灵活的，跟Spark一样，支持Local，Standalone，Yarn，Mesos，Kubernetes

代码中好像不能指定部署方式，和spark不同

https://blog.csdn.net/qq_33689414/article/details/90671685

1 Local

最简单的启动方式，其实是不搭建集群，直接本地启动。本地部署非常简单，直接解压安装包就可以使用，不用进行任何配置；一般用来做一些简单的测试。

2 Standalone

会话模式，应用模式

区别在于jobmaster的启动时间点，会话预先启动，应用在作业提交启动

3 Yarn

1 会话 session

在会话模式下，我们需要先启动一个YARN session，这个会话会创建一个 Flink集群。

2 单作业 per-job

flink不会预先启动，在提交作业，才启动新的jobmanager

3 应用application

与单作业很相似

区别在于提交给yarn资源管理器的不是具体作业，而是整个应用（包含了多个作业）

4 Mesos

5 Kubernetes

大数据基础组件 flink 原理

flink部署

2022-03-12 3c6c49eb8ded8cb735cb9466d4c40c6f 99+ fast 0.0 k

Flink架构原理

1.Flink组件

1 JobManager

2 任务管理器（ TaskManager）

2.任务调度流程

大数据基础组件 flink 原理

Flink架构原理

2022-03-12 8a09dcb69c6902cdb625146d82a3bffd 99+ a minute 0.1 k

flink vs spark

数据模型
1 spark 采用 RDD 模型， spark streaming 的 DStream 实际上也就是一组组小批
数据 RDD 的集合
2 flink 基本数据模型是数据流，以及事件（ Event ）序列
运行时架构
1 spark 是批计算，将 DAG 划分为不同的 stage ，一个完成后才可以计算下一个
2 flink 是标准的流执行模式，一个事件在一个节点处理完后可以直接发往下一个节
点进行处理

大数据基础组件 flink 原理

flink vs spark

1 source

1 从集合中读取数据

2 从文件读取数据

3 socket

4 kafka

5 自定义 Source

3 sink

1 输出到文件

2 输出到 Kafka

3 输出到 Redis

4 输出到 Elasticsearch

5 输出到 MySQL (JDBC)

6 自定义 Sink输出

性能

task

worker

executor

内存

driver

executor

资源参数参考示例

1 Local

2 Standalone

3 Yarn

4 Mesos

5 Kubernetes

1.Flink组件

1 JobManager

2 任务管理器（ TaskManager）

2.任务调度流程

Recents

Categories

Archives

Tags

Subscribe for updates