Flink知识点

1. Flink、Storm、Sparkstreaming对比

Storm只支持流处理任务，数据是一条一条的源源不断地处理，而MapReduce、spark只支持批处理任务，spark-streaming本质上是一个批处理，采用micro-batch的方式，将数据流切分成细粒度的batch进行处理。Flink同时支持流处理和批处理，一条数据被处理完以后，序列化到缓存后，以固定的缓存块为单位进行网络数据传输，缓存块设为0为流处理，缓存块设为较大值为批处理。

storm------ ---------At-least-once----Record Acks-------无状态管理-------- 低延迟-----------高吞吐

sparkstreaming-----Exactly-once-------RDD Checkpoint-----基于DStream-----中等延迟-----高吞吐

Flink-----------------Exactly-once------ Checkpoint-----------基于操作----------低延迟-------高吞吐

2. Flink DataStream API基本算子

map:做一些清洗转换；

flatMap:输入一个元素，返回一个或者多个元素；

filter:符合条件的数据留下；

keyBy:key相同的数据进入同一个分区；

reduce:当前元素与上一次reduce返回值进行聚合操作；

Union:合并多个流，但是所有的流类型必须一致；

Connect:合并两个流，两个流的类型可以不同；

CoMap、CoFlatMap:对于ConnectedStream使用这俩函数，对两个流进行不同的处理；

split:根据规则吧一个数据流切分为多个流；

Select：配合split，选择切分后的流；

3. Flink On Yarn

Flink on yarn上的好处：

• 提高集群机器的利用率；

• 一套集群，可以同时执行MR任务，Spark任务，Flink任务。

Flink on yarn的两种方式：

• 一开始在Yarn上初始化一个集群；yarn-session.sh【开辟资源】+flink run【提交任务】

• 每个Flink job都申请一个集群，互不影响，任务执行之后资源会被释放掉。flink run -m yarn-cluster【开辟资源+提交任务】

4. FlinkKafkaConnector（FlinkKafkaConsumer+FlinkKafkaProducer）

阅读过FlinkKafkaConnector源码后对Kafka偏移量的存储机制有了一个全新的认识，这里有两个坑，简单记下：

• FlinkKafkaComsumer08用的Kafka老版本Consumer，偏移量提交Zookeeper，由Zookeeper保管，而FlinkKafkaComsumer09以后，偏移量默认存在Kafka内部的Topic中，不再向Zookeeper提交；
• 使用FlinkKafkaProducer08写Kafka1.10（其他高版本没试过）存在超时。

5. DataStream API之partition

• 随机分区

 dataStream.shuffle();

• 重分区，消除数据倾斜

 dataStream.rebalance();

• 自定义分区

 dataStream.partitionCustom(partiitoner,"somekey");

• 广播分区：把元素广播所有分区，会被重复消费

 DataStream.broadcast();

6. Flink Distributed Cache(分布式缓存)

• 原理

类似Hadoop，可以在并行函数比如map中读取本地文件或则HDFS文件，Flink自动将文件复制到所有taskmanager节点的本地文件系统。

• 用法

用法一：注册一个文件

 env.registerCachedFile(“hdfs:///path/to/yout/file","hdfsfile");

用法二：访问数据

 File myFile = getRuntimeContext().getDistributedCache().getFile("hdfsFile");

7. State

为了实现at least once和exactly once，flink引入了state和checkpoint；state一般将数据保存在堆内存中，而checkpoint是每隔一段时把state数据持久化存储了，当失败时可以恢复。

8. Checkpoint

依赖checkpoint机制，只能保证Flink系统内的exactly once。

checkpoint默认是disabled的，开启之后默认是exactly once，这种模式对大多数应用合适，而at least once在某些低延迟的场景中比较合适（例如几毫秒）。例如，

 env.enableCheckpointing(1000);                                            //每隔1000ms设置一个检查点【检查点周期】
 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);//设置模式为exactly once（默认值）
 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);//检查点之间至少有500ms的间隔【检查点最小间隔】
 env.getCheckpointConfig().setCheckpointTimeout(60000); //检查点必须在一分钟之内完成，否则丢弃【检查点的超时时间】
 env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); //同一时间点只允许一个检查点
6 env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
7 //Flink程序被取消后，会保留检查点数据                

9. Flink重启策略

Flink中常用重启策略：

固定时间（Fixed delay）;

失败率（Failure rate）【checkpoint机制默认策略】;

无重启（No start）【默认】

10. Window（窗口）

Window将一个stream拆分成有限大小的"桶（buckets）"，在这些桶上做计算。Flink中的窗口与可以分为时间驱动【Time Window】和数据驱动【Count Window】，每一个可以细分为：

• 滚动窗口【没有重叠】tumbling windows

 .timeWindow(Time.minutes(1));
 .countWindow(100);

• 滑动窗口【有重叠】sliding windows

 .timeWindow(Time.minutes(1),Time.seconds(30))//每隔30秒，统计一分钟的数据
 .countWindow(100,10)//每隔10条数据，统计100条的数据

• 会话窗口 session windows

定义了窗口之后，会存在不同的聚合：

• 增量聚合（窗口中每进入一条数据，就进行一次计算）,例如reduce、aggregate、sum、min、max；

• 全量聚合（等窗口所有数据到齐，才开始计算，可以进行排序等）例如apply、process等

当然也可以分为keyed window和non-keyed window，分组的stream调用keyBy(...)和window(...)，非分组的stream中window(...)换成了windowAll(...)

11. Time

针对Stream数据流中的时间，可以分为以下三种：

• Event Time：日志产生的时间；

• Ingestion Time：事件从kafka等取出来，进入Flink的时间；

• Processing Time：事件被处理的时间，例如达到窗口处理的时间等。【默认】

12. Flink并行度

• Flink中每个TaskManager为集群提供slot，slot数量与每个节点的可用CPU核数成比例，slot上启动进程，进程内有多个线程。如果任务管理器有n个槽，它会为每个槽分配 1/n 的内存，这里没有对 CPU 进行隔离；目前任务槽仅仅用于划分任务的内存。

配置一个TaskManager有多少个并发的slot数有两种配置方式：

• taskmanager.numberOfTaskSlots。在conf/flink-conf.yaml中更改，默认值为1，表示默认一个TaskManager只有1个task slot.
• 提交作业时通过参数配置。--yarnslots 1，表示TaskManager的slot数为1.

⚠️注意：slot不能搞太多，几十个就行，你想啊假如你机器不多，TaskManager不多，搞那么多slot，每个slot分到的内存小的可怜，容易OOM啊

• 并行度的设置有多个地方：操作算子层面、执行环境层面、客户端层面、系统层面，具体可以参考：FLINK并行度；

算子层面具体情况具体分析，例如KafkaSource的并行度最好跟Kafka Topic的分区数成比例，比如：

 val wordCounts = text
    .flatMap{ _.split(" ") map { (_, 1) } }
    .keyBy(0)
    .timeWindow(Time.seconds(5))
    .sum(1).setParallelism(5)

执行环境层面比如：

 env.setParallelism(5)

提交任务的时候，在客户端侧flink可以通过-p参数来设置并行度。例如：

 ./bin/flink run -p 5 ../examples/*WordCount-java*.jar

13. TaskManager数量

TaskManager的数量是在提交作业时根据并行度动态计算。先根据设定的operator的最大并行度计算，例如，如果作业中operator的最大并行度为10，则 Parallelism/numberOfTaskSlots为向YARN申请的TaskManager数。

14. Flink传递参数给函数

参数可以使用构造函数或withParameters(Configuration)方法传递，参数将会作为函数对象的一部分被序列化并传递到task实例中。

• 使用构造函数方式

 DataSet toFilter = env.fromElements(1, 2, 3);
 toFilter.filter(new MyFilter(2));
 private static class MyFilter implements FilterFunction {
   private final int limit;
   public MyFilter(int limit) {
     this.limit = limit;
   }
   @Override
   public boolean filter(Integer value) throws Exception {
     return value > limit;
   }
 }

（2）withParameters(Configuration)方式

这个方法将携带一个Configuration对象作为参数，参数将会传递给Rich Function的open方法(关于Rich Function参见：rich function)。Configuration对象是一个Map，存储Key/Value键值对.

 DataSet toFilter = env.fromElements(1, 2, 3);
 Configuration config = new Configuration();
 config.setInteger("limit", 2);
 toFilter.filter(new RichFilterFunction() {
     private int limit;
     @Override
     public void open(Configuration parameters) throws Exception {
       limit = parameters.getInteger("limit", 0);
     }
     @Override
     public boolean filter(Integer value) throws Exception {
       return value > limit;
     }
 }).withParameters(config);

（3）使用全局的the ExecutionConfig方式

参数可以被所有的rich function获得

 Configuration conf = new Configuration();
 conf.setString("mykey","myvalue");
 final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
 env.getConfig().setGlobalJobParameters(conf);
 public static final class Tokenizer extends RichFlatMapFunction> {
     private String mykey;
     @Override
     public void open(Configuration parameters) throws Exception {
       super.open(parameters);
       ExecutionConfig.GlobalJobParameters globalParams = getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
       Configuration globConf = (Configuration) globalParams;
       mykey = globConf.getString("mykey", null);
     }
     // ... more here ...

15. Flink中的Metrics

Flink中有丰富的Metrics指标，当然我们也可以使用它的Reporter自定义，具体参见官网，示例如下：

 import org.apache.flink.api.common.accumulators.LongCounter;
 import org.apache.flink.api.common.functions.RichMapFunction;
 import org.apache.flink.api.java.utils.ParameterTool;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.dropwizard.metrics.DropwizardMeterWrapper;
 import org.apache.flink.metrics.Counter;
 import org.apache.flink.metrics.Meter;
 import org.apache.flink.metrics.MetricGroup;

 public class Map extends RichMapFunction<String, String> {
   private transient Meter logTotalMeter;
   private transient Counter logTotalConter;
   private LongCounter logAcc = new LongCounter();

   @Override
   public void open(Configuration parameters) throws Exception {
     MetricGroup metricGroup = getRuntimeContext().getMetricGroup();
     logTotalMeter = metricGroup.meter("logTotalMeter",
         new DropwizardMeterWrapper(new com.codahale.metrics.Meter()));
     logTotalConter = metricGroup.counter("logTotalConter");
     getRuntimeContext().addAccumulator("logAcc", this.logAcc);
   }

   @Override
   public String map(String log) throws Exception {
     logTotalConter.inc();
     logTotalMeter.markEvent();
     logAcc.add(NumberUtil.LONG_ONE);
     logMeter.mark();
     return log;
   }
 }

16. Flink中的table API

听了阿里军长的讲解后有所了解，简单做下笔记，以免忘记。table API和SQL有很多相同的优良特性，比如：

声明式-用户只关心做什么，不用关心怎么做；

高性能-支持查询优化，可以获取更好的执行性能；

流批统一-相同的统计逻辑，既可以流模式也可以批模式；

标准稳定-语义遵循SQL标准，不易变动

易理解-所见即所得

比如：

 // table API
 tab.groupBy("word").select("word,count(1) as cnt")

 // SQL
 SELECT COUNT(1) AS cnt
 FROM tab
 GROUP BY word

总的来说，SQL有的功能table API都有，如下图所示

 环境配置如下

三种注册表的方式

三种发射表的方式

table API对列操作比较方便，比如

再比如有一张100列的表，选择1到10列怎么操作？

另外，table API的map函数扩展起来也很方便

也可参考：Table API & SQL、Flink SQL-Client。

17. Flink中的内存管理

写这个也是有点心累的，搞了好久，也算对Flink Web UI上的几个内存指标大致了解了，老规矩记录下。所有程序都是绝对受人控制的，在提交任务那一刻，我们可以指定程序中使用的内存，例如：

 ./bin/flink run -m yarn-cluster -yn 2 -yjm 1024 -ytm 1024 ./examples/batch/WordCount.jar

红色参数的具体含义是（参见Flink官网）：

-yjm,--yarnjobManagerMemory <arg>    Memory for JobManager Container
                                          with optional unit (default: MB)
-ytm,--yarntaskManagerMemory <arg>   Memory per TaskManager Container
                                          with optional unit (default: MB)

先上两个小图（这俩图我没找到原版出处，一直觉得这玩意坑爹，大概率上这里的JVM Heap包括了on-heap和off-heap，on-heap和off-heap下边有介绍）

看图说话，TaskManager 的堆内存主要被分成了三个部分：

• Network Buffers: 一定数量的32KB大小的 buffer，主要用于数据的网络传输。在 TaskManager 启动的时候就会分配。默认数量是 2048 个，可以通过 taskmanager.network.numberOfBuffers来配置。
• Memory Manager Pool: 这是一个由 MemoryManager 管理的，由众多MemorySegment组成的超大集合。Flink 中的算法（如 sort/shuffle/join）会向这个内存池申请 MemorySegment，将序列化后的数据存于其中，使用完后释放回内存池。默认情况下，池子占了堆内存的 70% 的大小。
• Remaining (Free) Heap: 这部分的内存是留给用户代码以及 TaskManager 的数据结构使用的。因为这些数据结构一般都很小，所以基本上这些内存都是给用户代码使用的。从GC的角度来看，可以把这里看成的新生代，也就是说这里主要都是由用户代码生成的短期对象。

⚠️注意：Memory Manager Pool 主要在Batch模式下使用。在Steaming模式下，该池子不会预分配内存，也不会向该池子请求内存块。也就是说该部分的内存都是可以给用户代码使用的。不过社区是打算在 Streaming 模式下也能将该池子利用起来。

从堆的角度来说，Flink当前的内存支持堆内（on-heap）和堆外（off-heap）管理，用户想去申请什么类型的内存，有相关的参数去配置。Flink off-heap的内存管理相对于on-heap的优点主要在于：

• 启动分配了大内存(例如100G)的JVM很耗费时间，垃圾回收也很慢。如果采用off-heap，剩下的Network buffer和Remaining heap都会很小，垃圾回收也不用考虑MemorySegment中的Java对象了，节省了GC时间；
• 有效防止OOM，MemorySegment大小固定，操作高效。如果MemorySegment不足写到磁盘，内存中的数据不多，一般不会发生OOM；
• 更有效率的IO操作。在off-heap下，将MemorySegment写到磁盘或是网络可以支持zeor-copy技术，而on-heap的话则至少需要一次内存拷贝；
• off-heap上的数据可以和其他程序共享。

好了概念讲了一通，看下任务TaskManager日志

 2019-07-23 07:27:50,035 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -  Starting YARN TaskExecutor runner (Version: 1.7.1, Rev:<unknown>, Date:<unknown>)
 2019-07-23 07:27:50,035 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -  OS current user: yarn
 2019-07-23 07:27:50,436 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -  Current Hadoop/Kerberos user: worker
 2019-07-23 07:27:50,436 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -  JVM: Java HotSpot(TM) 64-Bit Server VM - Oracle Corporation - 1.8/25.112-b15
 2019-07-23 07:27:50,436 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -  Maximum heap size:  MiBytes
 2019-07-23 07:27:50,437 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -  JAVA_HOME: /usr/local/jdk/
 2019-07-23 07:27:50,438 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -  Hadoop version: 2.6.0-cdh5.5.0
 2019-07-23 07:27:50,438 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -  JVM Options:
 2019-07-23 07:27:50,438 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -     -Xmsm
 2019-07-23 07:27:50,439 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -     -Xmxm
 2019-07-23 07:27:50,439 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -     -XX:MaxDirectMemorySize=m

红色的地方有几个值，Flink源码中这几个值的计算在TaskManagerServices.calculateHeapSizeMB（计算堆内内存大小）和calculateNetworkBufferMemory（计算堆外内存大小），下边我们来看默认情况下这些值怎么计算的（仅限Flink 1.7.1版本，其他版本系数可能有变化）。

1. JVM预留内存，总内存的25%，最小预留，600M：

1024MB - (1024 * 0.25 < 600MB) -> 600MB = 424MB (cutoff)

2. 剩下的内存的10%作为networkBuffer的内存，最小64M：

424MB - (424MB * 0.1 < 64MB) -> 64MB(networkbuffer) = MB

3. 批作业的话剩下内存30%设为堆内内存，总内存减去堆内内存设为directMemory，流作业全部都是堆内内存了，用于netty和rocksDB和networkBuffer以及JVM自身内存。

这样360M和664M都明了了，那为啥一开始日志打印“Maximum heap size: 345 MiBytes”呢，这里还有一个小坑，先看Flink怎么计算的，

     /**
      * The maximum JVM heap size, in bytes.
      *
      * <p>This method uses the <i>-Xmx</i> value of the JVM, if set. If not set, it returns (as
      * a heuristic) 1/4th of the physical memory size.
      *
      * @return The maximum JVM heap size, in bytes.
      */
     public static long getMaxJvmHeapMemory() {
         final long maxMemory = Runtime.getRuntime().maxMemory();
         if (maxMemory != Long.MAX_VALUE) {
             // we have the proper max memory
             return maxMemory;
         } else {
             // max JVM heap size is not set - use the heuristic to use 1/4th of the physical memory
             final long physicalMemory = Hardware.getSizeOfPhysicalMemory();
             if (physicalMemory != -1) {
                 // got proper value for physical memory
                 return physicalMemory / 4;
             } else {
                 throw new RuntimeException("Could not determine the amount of free memory.\n" +
                         "Please set the maximum memory for the JVM, e.g. -Xmx512M for 512 megabytes.");
             }
         }
     }

它用了Java lang的方法，这就有问题了，问题在于JVM使用的GC算法都会有一些内存丢失，比如Survivor有两个，但只有1个会用到，另一个一直闲置，总有一块Survivor区是不被计算到可用内存中的。

到底是不是呢，我在Flink程序的Map函数中加了这么一段代码：

 logger.info("Runtime max: " + mb(Runtime.getRuntime().maxMemory()));
 MemoryMXBean m = ManagementFactory.getMemoryMXBean();

 logger.info("Non-heap: " + mb(m.getNonHeapMemoryUsage().getMax()));
 logger.info("Heap: " + mb(m.getHeapMemoryUsage().getMax()));

 for (MemoryPoolMXBean mp : ManagementFactory.getMemoryPoolMXBeans()) {
   logger.info("Pool: " + mp.getName() + " (type " + mp.getType() + ")" + " = " + mb(mp.getUsage().getMax()));
 }

打印日志出来

 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Heap: 361758720 (345.00 M)
 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Pool: PS Survivor Space (type Heap memory) = 15728640 (15.00 M)
 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Pool: Code Cache (type Non-heap memory) = 251658240 (240.00 M)
 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Pool: PS Old Gen (type Heap memory) = 251658240 (240.00 M)
 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Pool: Metaspace (type Non-heap memory) = -1 (-0.00 M)
 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Pool: Compressed Class Space (type Non-heap memory) = 1073741824 (1024.00 M)
 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Pool: PS Eden Space (type Heap memory) = 94371840 (90.00 M)
 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Pool: PS Survivor Space (type Heap memory) =  (15.00 M)
 2019-07-20 09:03:16,345 INFO  data.demo.core.example                             - Pool: PS Old Gen (type Heap memory) = 251658240 (240.00 M)
 2019-07-20 09:03:16,345 INFO  data.demo.core.example                             - Runtime max: 361758720 (345.00 M)

眼见为实，实锤了。

再看Flink TaskManager WebUI

有一些概念还要说下，比如JVM管理两种类型的内存：堆（heap）和非堆（Nonheap），堆就是Java代码可及的内存，所有类实例和数组的内存都是在堆上分配。非堆就是JVM留给自己用的，方法区、栈、每个类结构(如运行时常数池、字段和方法数据)以及方法和构造方法的代码都在非堆内存中。

再比如Direct和Mapped，这俩是JVM缓冲池，主要是JNI使用。

UI上的这些参数我们都可以通过Flink的Rest API拿到下面这些指标值，比如通过访问本地localhost:23799/taskmanagers/container_e37_1563420494990_0370_01_000009（具体参见Flink官网）：

Scope	Infix	Metrics	Description
Job-/TaskManager	Status.JVM.Memory	Memory.Heap.Used	当前使用的堆内存大小.
		Heap.Committed	保证JVM可用的堆内存大小.
		Heap.Max	可用于内存管理的堆内存最大值.
		NonHeap.Used	当前使用的非堆内存大小.
		NonHeap.Committed	保证JVM可用的非堆内存大小.
		NonHeap.Max	可用于内存管理的非堆内存最大值.
		Direct.Count	直接缓冲池中的缓冲区数量.
		Direct.MemoryUsed	JVM中用于直接缓冲池的内存大小.
		Direct.TotalCapacity	直接缓冲池中所有缓冲区的总容量.
		Mapped.Count	映射缓冲池中缓冲区的数量.
		Mapped.MemoryUsed	JVM中用于映射缓冲池的内存大小.
		Mapped.TotalCapacity	映射缓冲池中缓冲区的数量.

18. Flink中依赖配置

    每个Flink应用程序都依赖一组Flink库，至少应用程序依赖于Flink API。许多应用程序还依赖于某些连接器库（如Kafka，Cassandra等）。所以阅读了下官网依赖配置这一节的内容，简单记一下，Flink中有两大类依赖项和库：

• Flink核心依赖：Flink本身由运行系统所需的一组类和依赖项组成，例如协调，网络，检查点，故障转移，API，操作（如窗口），资源管理等。所有这些这些类和依赖项构成了Flink运行时的核心，在启动Flink应用程序时必须存在。这些核心类和依赖项打包在flink-dist.jar中。它们是Flink lib文件夹的一部分。这部分不包含任何连接器或库（CEP，SQL，ML等），以避免默认情况下在类路径中具有过多的依赖项和类，保持默认的类路径较小并避免依赖性冲突。Maven（和其他构建工具）将依赖项打包时一般将核心依赖设为provided，如果它们未设置为provided，可能使生成的JAR包过大，还可能出现添加到应用程序的jar文件的Flink核心依赖项与用户自己的一些依赖版本冲突（可以通过反向类加载来避免）；如果在IntelliJ IDEA中调试，则将scope设置为comiple，否则失败报NoClassDefFountError错；

• 用户应用程序依赖: connectors, formats, or libraries(CEP, SQL, ML)，用户应用程序通常打包到应用程序jar中。

另外，当Flink程序读写HDFS时需要添加Hadoop依赖，不要把Hadoop依赖直接添加到Flink application，而是： export HADOOP_CLASSPATH=`hadoop classpath`，Flink组件启动时会使用该环境变量，这样做是因为：

• 一些Hadoop交互发生在Flink的核心，可能在用户应用程序启动之前，例如为检查点设置HDFS，通过Hadoop的Kerberos令牌进行身份验证或在YARN上部署。
• Flink的反向类加载方法隐藏了核心依赖关系中的许多传递依赖关系，应用程序可以使用相同依赖项的不同版本，而不会遇到依赖项冲突。

19. Flink单元测试指南

    参考：指南文档翻译

20. Flink读HDFS

    简单点说，读数据

 DataSet<String> hdfslines=env.readTextFile("your hdfs path")

写数据

 hdfslines.writeAsText("your hdfs path")

以上会根据你的默认的线程数来生成多少个分区文件，如果你想最后生成一个文件的话，可以在后面使用setParallelism(1)，这样最后就只会生成一个文件了。具体可以这么整

    try {
       String topic = args[0];
       String path = args[1];
       //读取配置文件
       Configuration conf = new Configuration();
       //获取文件系统
       FileSystem fs = FileSystem.get(URI.create("/"), conf);

       System.out.println(fs.getUri());
       if (!fs.getUri().toString().contains("hdfs")) {
         path = "hdfs://localhost:8020" + path;
       }

       ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
       DataSet<String> union = env.readTextFile(fs.getUri() + path + "/").setParallelism(20);

       union.rebalance().setParallelism(15).map(new MapFunction<String, String>() {
         private static final long serialVersionUID = 1033071381217373267L;

         @Override
         public String map(String rawLog) throws Exception {
           return rawLog;
         }
       }).output(new DiscardingOutputFormat<>())
           .setParallelism(20);
       try {
         env.execute();
       } catch (Exception e) {
         e.printStackTrace();
       }

       //关闭文件系统
       fs.close();
     } catch (Exception e) {
       logger.error("task submit process error, due to {}.", e.getMessage());
       e.printStackTrace();
     }

path是传入的目录，比如/user/rawlog/hourly/2019-09-15/09

配置

21. 

    https://www.cnblogs.com/ljygz/p/11807064.html

    https://www.cnblogs.com/ljygz/p/11837033.html