Flink  内存

标签: flink

 Flink 

map:做一些清洗转换;

flatMap:输入一个元素,返回一个或者多个元素;

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

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

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

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

Connect:合并两个流,两个流的类型可以不同;

CoMapCoFlatMap:对于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

  • 随机分区
1 dataStream.shuffle();
  • 重分区,消除数据倾斜
1 dataStream.rebalance();
  • 自定义分区
1 dataStream.partitionCustom(partiitoner,"somekey");
  • 广播分区:把元素广播所有分区,会被重复消费
1 DataStream.broadcast();

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

  • 原理

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

  • 用法

用法一:注册一个文件

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

用法二:访问数据

1 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在某些低延迟的场景中比较合适(例如几毫秒)。例如,

复制代码

1 env.enableCheckpointing(1000);                                            //每隔1000ms设置一个检查点【检查点周期】             
2 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);//设置模式为exactly once(默认值)                 
3 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);//检查点之间至少有500ms的间隔【检查点最小间隔】        
4 env.getCheckpointConfig().setCheckpointTimeout(60000); //检查点必须在一分钟之内完成,否则丢弃【检查点的超时时间】
5 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
1 .timeWindow(Time.minutes(1));
2 .countWindow(100);
  • 滑动窗口【有重叠】sliding windows
1 .timeWindow(Time.minutes(1),Time.seconds(30))//每隔30秒,统计一分钟的数据
2 .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的分区数成比例,比如:

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

执行环境层面比如:

1 env.setParallelism(5)

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

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

13. TaskManager数量

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

14. Flink传递参数给函数

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

  • 使用构造函数方式

复制代码

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

复制代码

(2)withParameters(Configuration)方式

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

复制代码

 1 DataSet toFilter = env.fromElements(1, 2, 3);
 2 Configuration config = new Configuration();
 3 config.setInteger("limit", 2);
 4 toFilter.filter(new RichFilterFunction() {
 5     private int limit;
 6     @Override
 7     public void open(Configuration parameters) throws Exception {
 8       limit = parameters.getInteger("limit", 0);
 9     }
10     @Override
11     public boolean filter(Integer value) throws Exception {
12       return value > limit;
13     }
14 }).withParameters(config);

复制代码

(3)使用全局的the ExecutionConfig方式

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

复制代码

 1 Configuration conf = new Configuration();
 2 conf.setString("mykey","myvalue");
 3 final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
 4 env.getConfig().setGlobalJobParameters(conf);
 5 public static final class Tokenizer extends RichFlatMapFunction> {
 6     private String mykey;
 7     @Override
 8     public void open(Configuration parameters) throws Exception {
 9       super.open(parameters);
10       ExecutionConfig.GlobalJobParameters globalParams = getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
11       Configuration globConf = (Configuration) globalParams;
12       mykey = globConf.getString("mykey", null);
13     }
14     // ... more here ...

复制代码

15. Flink中的Metrics

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

复制代码

 1 import org.apache.flink.api.common.accumulators.LongCounter;
 2 import org.apache.flink.api.common.functions.RichMapFunction;
 3 import org.apache.flink.api.java.utils.ParameterTool;
 4 import org.apache.flink.configuration.Configuration;
 5 import org.apache.flink.dropwizard.metrics.DropwizardMeterWrapper;
 6 import org.apache.flink.metrics.Counter;
 7 import org.apache.flink.metrics.Meter;
 8 import org.apache.flink.metrics.MetricGroup;
 9 
10 
11 public class Map extends RichMapFunction<String, String> {
12   private transient Meter logTotalMeter;
13   private transient Counter logTotalConter;
14   private LongCounter logAcc = new LongCounter();
15 
16   @Override
17   public void open(Configuration parameters) throws Exception {
18     MetricGroup metricGroup = getRuntimeContext().getMetricGroup();
19     logTotalMeter = metricGroup.meter("logTotalMeter",
20         new DropwizardMeterWrapper(new com.codahale.metrics.Meter()));
21     logTotalConter = metricGroup.counter("logTotalConter");
22     getRuntimeContext().addAccumulator("logAcc", this.logAcc);
23   }
24 
25 
26   @Override
27   public String map(String log) throws Exception {
28     logTotalConter.inc();
29     logTotalMeter.markEvent();
30     logAcc.add(NumberUtil.LONG_ONE);
31     logMeter.mark();
32     return log;
33   }
34 }

复制代码

16. Flink中的table API

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

声明式-用户只关心做什么,不用关心怎么做;

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

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

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

易理解-所见即所得

比如:

复制代码

1 // table API
2 tab.groupBy("word").select("word,count(1) as cnt")
3 
4 // SQL
5 SELECT COUNT(1) AS cnt
6 FROM tab
7 GROUP BY word

复制代码

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

 

 环境配置如下

 

三种注册表的方式

三种发射表的方式

 

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

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

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

 

 也可参考:Table API & SQLFlink SQL-Client

17. Flink中的内存管理

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

1 ./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日志

复制代码

 1 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>)
 2 2019-07-23 07:27:50,035 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -  OS current user: yarn
 3 2019-07-23 07:27:50,436 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -  Current Hadoop/Kerberos user: worker
 4 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
 5 2019-07-23 07:27:50,436 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -  Maximum heap size: 345 MiBytes
 6 2019-07-23 07:27:50,437 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -  JAVA_HOME: /usr/local/jdk/
 7 2019-07-23 07:27:50,438 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -  Hadoop version: 2.6.0-cdh5.5.0
 8 2019-07-23 07:27:50,438 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -  JVM Options:
 9 2019-07-23 07:27:50,438 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -     -Xms360m
10 2019-07-23 07:27:50,439 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -     -Xmx360m
11 2019-07-23 07:27:50,439 INFO  org.apache.flink.yarn.YarnTaskExecutorRunner                  -     -XX:MaxDirectMemorySize=664m

复制代码

红色的地方有几个值,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) = 360MB 

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

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

复制代码

 1     /**
 2      * The maximum JVM heap size, in bytes.
 3      * 
 4      * <p>This method uses the <i>-Xmx</i> value of the JVM, if set. If not set, it returns (as
 5      * a heuristic) 1/4th of the physical memory size.
 6      * 
 7      * @return The maximum JVM heap size, in bytes.
 8      */
 9     public static long getMaxJvmHeapMemory() {
10         final long maxMemory = Runtime.getRuntime().maxMemory();
11         if (maxMemory != Long.MAX_VALUE) {
12             // we have the proper max memory
13             return maxMemory;
14         } else {
15             // max JVM heap size is not set - use the heuristic to use 1/4th of the physical memory
16             final long physicalMemory = Hardware.getSizeOfPhysicalMemory();
17             if (physicalMemory != -1) {
18                 // got proper value for physical memory
19                 return physicalMemory / 4;
20             } else {
21                 throw new RuntimeException("Could not determine the amount of free memory.\n" +
22                         "Please set the maximum memory for the JVM, e.g. -Xmx512M for 512 megabytes.");
23             }
24         }
25     }

复制代码

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

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

复制代码

1 logger.info("Runtime max: " + mb(Runtime.getRuntime().maxMemory()));
2 MemoryMXBean m = ManagementFactory.getMemoryMXBean();
3 
4 logger.info("Non-heap: " + mb(m.getNonHeapMemoryUsage().getMax()));
5 logger.info("Heap: " + mb(m.getHeapMemoryUsage().getMax()));
6 
7 for (MemoryPoolMXBean mp : ManagementFactory.getMemoryPoolMXBeans()) {
8   logger.info("Pool: " + mp.getName() + " (type " + mp.getType() + ")" + " = " + mb(mp.getUsage().getMax()));
9 }

复制代码

打印日志出来

复制代码

 1 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Heap: 361758720 (345.00 M)
 2 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Pool: PS Survivor Space (type Heap memory) = 15728640 (15.00 M)
 3 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Pool: Code Cache (type Non-heap memory) = 251658240 (240.00 M)
 4 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Pool: PS Old Gen (type Heap memory) = 251658240 (240.00 M)
 5 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Pool: Metaspace (type Non-heap memory) = -1 (-0.00 M)
 6 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Pool: Compressed Class Space (type Non-heap memory) = 1073741824 (1024.00 M)
 7 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Pool: PS Eden Space (type Heap memory) = 94371840 (90.00 M)
 8 2019-07-20 09:03:16,344 INFO  data.demo.core.example                             - Pool: PS Survivor Space (type Heap memory) = 15728640 (15.00 M)
 9 2019-07-20 09:03:16,345 INFO  data.demo.core.example                             - Pool: PS Old Gen (type Heap memory) = 251658240 (240.00 M)
10 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

    简单点说,读数据

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

写数据

1 hdfslines.writeAsText("your hdfs path")

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

复制代码

 1    try {
 2       String topic = args[0];
 3       String path = args[1];
 4       //读取配置文件
 5       Configuration conf = new Configuration();
 6       //获取文件系统
 7       FileSystem fs = FileSystem.get(URI.create("/"), conf);
 8 
 9       System.out.println(fs.getUri());
10       if (!fs.getUri().toString().contains("hdfs")) {
11         path = "hdfs://localhost:8020" + path;
12       }
13 
14       ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
15       DataSet<String> union = env.readTextFile(fs.getUri() + path + "/").setParallelism(20);
16 
17       union.rebalance().setParallelism(15).map(new MapFunction<String, String>() {
18         private static final long serialVersionUID = 1033071381217373267L;
19 
20         @Override
21         public String map(String rawLog) throws Exception {
22           return rawLog;
23         }
24       }).output(new DiscardingOutputFormat<>())
25           .setParallelism(20);
26       try {
27         env.execute();
28       } catch (Exception e) {
29         e.printStackTrace();
30       }
31 
32       //关闭文件系统
33       fs.close();
34     } catch (Exception e) {
35       logger.error("task submit process error, due to {}.", e.getMessage());
36       e.printStackTrace();
37     }

复制代码

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

版权声明:本文为wangshuminjava原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://blog.csdn.net/wangshuminjava/article/details/107608469

智能推荐

模拟按键 —— 鼠标

背景 之前写自动化脚本的时候总是遇到一些很尴尬的问题: 跑脚本时模拟鼠标按键时,光标是真实的跑到了那个位置的,也就是说跑脚本的时候会很影响电脑的正常使用,导致不得不开一个虚拟机专门跑。 另外因为光标只有一个所以很难实现多线程去同时操作多个窗口,当线程1 模拟鼠标但还没有结束时,线程2 已经开始执行模拟操作,这就导致了线程1 的模拟操作被终止了,被迫之下只能开多个虚拟机(但实在太占用性能🙄) 解决...

Hibernate学习总结(一)

一、Hibernate简介 一个持久层的ORM框架。ORM:Object Relational Mapping(对象关系映射)。指的是将一个Java中的对象与关系型数据库中的表建立一种映射关系,从而操作对象就可以操作数据库中的表。 二、Hibernate入门 1、创建一个项目,引入jar包 hibernate用到的jar包 2、创建表 3、创建实体类 4、创建映射(*****) 映射需要通过XML...

Linux系统NFS

文章目录 1. nfs简介 1.1 nfs特点 1.2 使用nfs的好处 1.3 nfs的体系组成 1.4 nfs的应用场景 2. nfs工作机制 2.1 RPC 2.2 NIS 2.3 nfs工作机制 3. exports文件的格式 4. nfs管理 5. 作业 5.1手动搭建一个nfs服务器 5.1.1开放/nfs/shared目录,供所有用户查阅资料 5.1.2 开放/nfs/upload目...

关于java中String,StringBuffer,StringBuilder的区别以及StringBuffer,StringBuilder的安全性问题

这里的结果就是正确的然后我们来看他的append方法 它在前边加了一个synchronized来修饰,相当于同时只能有一个线程来访问他,这样就不会产生上边的问题但同时他的效率也就比StringBuilder低,...

Django连接现有mysql数据库

1、打开cmd后cd到项目位置 2、建立项目 django-admin startproject test2 3、编辑项目中的配置文件, mysite/settings.py ,告诉Django你的数据库连接参数和数据库名。具体的说,要提供 DATABASE_NAME , DATABASE_ENGINE , DATAB...

猜你喜欢

ShareSDK新浪微博登录时报错error:redirect_uri_mismatch

今天用 ShareSDK 做第三方登录的时候碰到个问题,明明在微博平台的应用审核已经通过了,但是调用登录接口的时候一直报错,错误如下: 出现这个错误是因为在微博开放平台上没有设置回调地址,或者设置的回调地址与本地XML中的地址不一致。 在sharesdk.xml文件当中对于微博的设置: 其中RedirectUrl为设置的回调地址,这里的地址必须要与微博开发平台设置的地址相同,否则就会出现上面的错误...

python解析网络封包方法

2019独角兽企业重金招聘Python工程师标准>>> 在使用Python解析网络数据包时,使用网络字节序解析,参见下表。 C语言的数据类型和Python的数据类型对照表请参见下表。 接下来对封包与解包进行举例说明。 version type id content unsigned short unsigned short unsigned int unsigned int 封包...

python3:时间方法,异常处理,系统文件相关模块(os)

文章目录 时间方法 time模块 时间表示方法: time模块的方法 datetime模块 异常处理 触发异常 创建mydiv.py脚本,要求如下: 创建myerror.py脚本,要求如下: os模块 实现ls -R(os.walk) os.path pickle模块 记账脚本 时间方法 time模块 时间表示方法: 时间戳:自1970-1-1 0:00:00到某一时间点之间的秒数 UTC时间:世...

负载均衡群集——LVS+DR模型

一、实验组成 调度器 192.168.100:41 web1 192.168.100:42 web2 192.168.100.43 NFS共享服务器 192.168.100.44 二、实验拓扑 三、实验配置 3.1在调度器配置:192.168.100.41 配置虚拟IP地址(VIP) 调整/proc响应参数 对于 DR 群集模式来说,由于 LVS 负载调度器和各节点需要共用 VIP 地址,应该关闭...

adb无线连接时appium找不到设备

问题描述 以前使用USB连接真机,运行appium时一直正常,连接参数如下: 最近为了方便,使用adb无线连接真机,adb版本为1.0.40,真机安卓版本10,连接后,通过adb devices能够查看到连接的设备: adb无线连接是正常的,但每次运行时appium都找不到无线连接的设备,陷入重启adb循环: 解决流程 1.因为是没找到设备,所以在appium连接参数中增加了"udid&...