首页 > 其他分享 >DataStream API（一）

DataStream API（一）

时间：2023-12-30 16:44:23浏览次数：29  
标签：DataStream StreamExecutionEnvironment Flink API flink env apache new

DataStream API（一）

Flink 有非常灵活的分层 API 设计，其中的核心层就是 DataStream/DataSet API。由于新版 本已经实现了流批一体， DataSet API 将被弃用，官方推荐统一使用 DataStream API 处理流数 据和批数据。由于内容较多，我们将会用几章的篇幅来做详细讲解，本章主要介绍基本的DataStream API 用法。 DataStream（数据流）本身是 Flink 中一个用来表示数据集合的类（Class），我们编写的Flink 代码其实就是基于这种数据类型的处理，所以这套核心 API 就以 DataStream 命名。对于 批处理和流处理，我们都可以用这同一套 API 来实现。 DataStream 在用法上有些类似于常规的 Java 集合，但又有所不同。我们在代码中往往并 不关心集合中具体的数据，而只是用 API 定义出一连串的操作来处理它们；这就叫作数据流 的“转换”（transformations）。

一个 Flink 程序，其实就是对 DataStream 的各种转换。具体来说，代码基本上都由以下几部分构成

⚫ 获取执行环境（execution environment）

⚫ 读取数据源（source）

⚫ 定义基于数据的转换操作（transformations）

⚫ 定义计算结果的输出位置（sink）

⚫ 触发程序执行（execute）

其中，获取环境和触发执行，都可以认为是针对执行环境的操作。所以本章我们就从执行环境、数据源（source）、转换操作（transformation）、输出（sink）四大部分，对常用的 DataStream API 做基本介绍。 

5.1 执行环境（Execution Environment） 

1. getExecutionEnvironment

最简单的方式，就是直接调用 getExecutionEnvironment 方法。它会根据当前运行的上下文 直接得到正确的结果：如果程序是独立运行的，就返回一个本地执行环境；如果是创建了 jar包，然后从命令行调用它并提交到集群执行，那么就返回集群的执行环境。也就是说，这个方法会根据当前运行的方式，自行决定该返回什么样的运行环境。

  1. StreamExecutionEnvironment env = 
  2. StreamExecutionEnvironment . getExecutionEnvironment () ;

2. createLocalEnvironment

这个方法返回一个本地执行环境。可以在调用时传入一个参数，指定默认的并行度；如果 不传入，则默认并行度就是本地的 CPU 核心数。

  1. StreamExecutionEnvironment localEnv = 
  2. StreamExecutionEnvironment . createLocalEnvironment () ;

3. createRemoteEnvironment

这个方法返回集群执行环境。需要在调用时指定 JobManager 的主机名和端口号，并指定 要在集群中运行的 Jar 包。

  1. StreamExecutionEnvironmentremoteEnv=StreamExecutionEnvironment
  2. .createRemoteEnvironment(
  3. "host",//JobManager主机名
  4. 1234,//JobManager进程端口号
  5. "path/to/jarFile.jar"//提交给JobManager的JAR包
  6. );

在获取到程序执行环境后，我们还可以对执行环境进行灵活的设置。比如可以全局设置程 序的并行度、禁用算子链，还可以定义程序的时间语义、配置容错机制。

5.1.2 执行模式(Execution Mode)  

  1. / / 批处理环境 
  2. ExecutionEnvironment batchEnv=ExecutionEnvironment.getExecutionEnvironment(); 
  3. / / 流处理环境 
  4. StreamExecutionEnvironment env = 
  5. StreamExecutionEnvironment . getExecutionEnvironment () ;

⚫ 流执行模式（STREAMING）

这是 DataStream API 最经典的模式，一般用于需要持续实时处理的无界数据流。默认情况下，程序使用的就是 STREAMING 执行模式。

⚫ 批执行模式（BATCH）

专门用于批处理的执行模式, 这种模式下， Flink 处理作业的方式类似于 MapReduce 框架。 对于不会持续计算的有界数据，我们用这种模式处理会更方便。

⚫ 自动模式（AUTOMATIC）

在这种模式下，将由程序根据输入数据源是否有界，来自动选择执行模式。

1. BATCH 模式的配置方法  

（ 1）通过命令行配置

bin/ flink run -Dexecution.runtime-mode=BATCH ...

在提交作业时，增加 execution.runtime-mode 参数，指定值为 BATCH。

（2）通过代码配置

  1. StreamExecutionEnvironment env = 
  2. StreamExecutionEnvironment . getExecutionEnvironment ( ) ; 
  3. env . setRuntimeMode (RuntimeExecutionMode . BATCH) ;

建议: 不要在代码中配置，而是使用命令行。这同设置并行度是类似的：在提交作业时指 定参数可以更加灵活，同一段应用程序写好之后，既可以用于批处理也可以用于流处理。而在 代码中硬编码（hard code）的方式可扩展性比较差。

2. 什么时候选择 BATCH 模式  

Flink 本身持有的就是流处理的世界观，即使是批量数据，也可以看作“有界流”来进行处理。所以 STREAMING 执行模式对于有界数据和无界数据都是有效的；而 BATCH模式仅能用于有界数据。

 说明：如果都是用STREAMING处理，可能有时候不太高效。

原因：是因为流处理会输出一些中间数据，但是批处理直接输出最终结果。

用 BATCH 模式处理批量数据，用 STREAMING模式处理流式数据。因为数据有界的时候，直接输出结果会更加高效；而当数据无界的时候, 我 们没得选择——只有 STREAMING 模式才能处理持续的数据流。

5.1.3 触发程序执行

有了执行环境，我们就可以构建程序的处理流程了：基于环境读取数据源，进而进行各种转换操作，最后输出结果到外部系统。

需要注意的是，写完输出（sink）操作并不代表程序已经结束。因为当 main()方法被调用 时，其实只是定义了作业的每个执行操作，然后添加到数据流图中；这时并没有真正处理数据 ——因为数据可能还没来。Flink 是由事件驱动的，只有等到数据到来，才会触发真正的计算， 这也被称为“延迟执行”或“懒执行”（lazy execution）。

所以我们需要显式地调用执行环境的 execute()方法，来触发程序执行。execute()方法将一 直等待作业完成，然后返回一个执行结果（JobExecutionResult）。

env.execute();

5.2 源算子（Source）

Flink 可以从各种来源获取数据，然后构建 DataStream 进行转换处理。一般将数据的输入 来源称为数据源(data source)，而读取数据的算子就是源算子（source operator）。所以，source就是我们整个处理程序的输入端。

Flink 代码中通用的添加 source 的方式，是调用执行环境的 addSource()方法：

DataStream<String> stream = env.addSource(...);

方法传入一个对象参数，需要实现 SourceFunction 接口；返回 DataStreamSource。这里的DataStreamSource 类继承自 SingleOutputStreamOperator 类，又进一步继承自 DataStream。所以 很明显，读取数据的 source 操作是一个算子，得到的是一个数据流（DataStream）。

1.从文件最终读取数据

2.从集合中读取数据

3.从元素中读取数据

4.从Socket文本流中读取数据

  1. mport org.apache.flink.api.common.serialization.SimpleStringSchema;
  2. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  4. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
  5.  
  6. import java.util.ArrayList;
  7. import java.util.Properties;
  8.  
  9. public class SourceTest {
  10.     public static void main(String[] args) throws Exception {
  11.         //创建执行环境
  12.         StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
  13.         //设置并行度为1
  14.         env.setParallelism(1);
  15.  
  16.         //1.从文件中读取数据
  17.         DataStreamSource<String> stream1=env.readTextFile("input/clicks.txt");
  18.  
  19.         //2.从集合中读取数据
  20.         ArrayList<Integer> nums=new ArrayList<>();
  21.         nums.add(2);
  22.         nums.add(5);
  23.         DataStreamSource<Integer> numStream = env.fromCollection(nums);
  24.  
  25.         ArrayList<Event> events=new ArrayList<>();
  26.         events.add(new Event("Mary","./home",1000L));
  27.         events.add(new Event("Boe","./cart",2000L));
  28.  
  29.         DataStreamSource<Event> stream2 = env.fromCollection(events);
  30.  
  31.         //3.从元素中读取数据
  32.         DataStreamSource<Event> stream3 = env.fromElements(
  33.                 new Event("Mary", "./home", 1000L),
  34.                 new Event("Boe", "./cart", 2000L)
  35.         );
  36.  
  37.         //4.从Socket文本流中读取
  38.         DataStreamSource<String> stream4 = env.socketTextStream("hadoop102", 7777);
  39.  
  40.         //5.从Kafka中读取数据
  41.  
  42.         Properties properties = new Properties();
  43.  
  44.         properties.setProperty("bootstrap.servers","hadoop102:9092");
  45.         properties.setProperty("group.id","consumer-group");
  46.         properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  47.         properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
  48.         properties.setProperty("auto.offset.reset","latest");
  49.  
  50.  
  51.         DataStreamSource<String> kafkaStream = env.addSource(new FlinkKafkaConsumer<String>("clicks", new SimpleStringSchema(), properties));
  52.  
  53.  
  54. /*
  55.         stream1.print("1");
  56.  
  57.         numStream.print("nums");
  58.  
  59.         stream2.print("2");
  60.  
  61.         stream3.print("3");
  62.  
  63.         stream4.print("4");
  64.  
  65.  */
  67.         kafkaStream.print("kafka");
  68.  
  69.         env.execute();
  70.  
  71.     }
  72. }

5.从kafka中读取数据（这里重点说明一下）

Kafka 作为分布式消息传输队列，是一个高吞吐、易于扩展的消息系统。而消息队列的传 输方式，恰恰和流处理是完全一致的。所以可以说 Kafka 和 Flink 天生一对，是当前处理流式 数据的双子星。在如今的实时流处理应用中，由 Kafka 进行数据的收集和传输，Flink 进行分析计算。

想要以 Kafka 作为数据源获取数据，我们只需要引入 Kafka 连接器的依赖。Flink 官 方提供的是一个通用的 Kafka 连接器，它会自动跟踪最新版本的 Kafka 客户端。

  1. <dependency> 
  2.  <groupId>org.apache.flink</groupId> 
  3.  <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId> 
  4.  <version>${flink.version}</version> 
  5. </dependency> 
  1. import org.apache.flink.api.common.serialization.SimpleStringSchema; 
  2. import org.apache.flink.streaming.api.datastream.DataStreamSource; 
  3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; 
  4. import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; 
  5.  
  6. import java.util.Properties; 
  7.  
  8. public class SourceKafkaTest { 
  9.  public static void main(String[] args) throws Exception { 
  10.  StreamExecutionEnvironment env = 
  11. StreamExecutionEnvironment.getExecutionEnvironment(); 
  12.  env.setParallelism(1); 
  13.  
  14.  Properties properties = new Properties(); 
  15.  properties.setProperty("bootstrap.servers", "hadoop102:9092"); 
  16.  properties.setProperty("group.id", "consumer-group"); 
  17.  properties.setProperty("key.deserializer", 
  18. "org.apache.kafka.common.serialization.StringDeserializer"); 
  19.  properties.setProperty("value.deserializer", 
  20. "org.apache.kafka.common.serialization.StringDeserializer"); 
  21.  properties.setProperty("auto.offset.reset", "latest"); 
  22.  
  23.  DataStreamSource<String> stream = env.addSource(new 
  24. FlinkKafkaConsumer<String>( 
  25.  "clicks", 
  26.  new SimpleStringSchema(), 
  27.  properties 
  28.  )); 
  29.  
  30.  stream.print("Kafka"); 
  31.  
  32.  env.execute(); 
  33.  } 
  34. }

要想从kafka中读取数据，首先要先启动zookeeper和kafka

  1. #启动单节点zookeeper
  2.  
  3. cd /opt/module/zookeeper
  4. ./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties
  5.  
  6. #启动单节点kafka
  7. cd /opt/module/kafka
  8. ./bin/kafka-server-start.sh -daemon ./config/server.properties

然后创建一个生产者，然后Flink运行代码，消费数据。

./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic clicks

然后就可以生产数据了。

5.2.6 自定义 Source

我们想要读取的数据源来自某个外部系统，而 flink 既没有预实现的方法、也没有提供连接器， 又该怎么办呢？那就只好自定义实现 SourceFunction 了。接下来我们创建一个自定义的数据源，实现 SourceFunction 接口。主要重写两个关键方法：run()和 cancel()。

⚫ run()方法：使用运行时上下文对象（SourceContext）向下游发送数据；

⚫ cancel()方法：通过标识位控制退出循环，来达到中断数据源的效果。

  1. import org.apache.flink.streaming.api.datastream.DataStreamSource; 
  2. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; 
  3.  
  4. public class SourceCustom { 
  5.  public static void main(String[] args) throws Exception { 
  6.  StreamExecutionEnvironment env = 
  7. StreamExecutionEnvironment.getExecutionEnvironment(); 
  8.  env.setParallelism(1); 
  9.  
  10.  //有了自定义的 source function，调用 addSource 方法 
  11.  DataStreamSource<Event> stream = env.addSource(new ClickSource()); 
  12.  
  13.  stream.print("SourceCustom"); 
  14.  
  15.  env.execute(); 
  16.  
  17.  } 
  18.  
  19.  
  20. }

这里要注意的是 SourceFunction 接口定义的数据源，并行度只能设置为 1，如果数据源设 置为大于 1 的并行度，则会抛出异常。 

 所以如果我们想要自定义并行的数据源的话，需要使用 ParallelSourceFunction，示例程序 如下

  1. package com.atguigu.chapter05;
  2.  
  3. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
  6.  
  7. import java.util.Random;
  8.  
  9. public class ParallelSourceExample {
  10.     //实现自定义并行的数据源
  11.     public static void main(String[] args) throws Exception {
  12.         StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
  13.         //先设置全局并行度为1
  14.         env.setParallelism(4);
  15.  
  16.         DataStreamSource<Integer> dataStreamSource=env.addSource(new ParallelSource()).setParallelism(2);
  17.  
  18.         dataStreamSource.print();
  19.  
  20.         env.execute();
  21.     }
  22. }
  23. class ParallelSource implements ParallelSourceFunction<Integer> {
  24.  
  25.     //声明一个标志位
  26.     private Boolean running=true;
  27.  
  28.     private Random random=new Random();
  29.  
  30.     public ParallelSource() {
  31.     }
  32.  
  33.     @Override
  34.     public void run(SourceContext<Integer> sourceContext) throws Exception {
  35.  
  36.         while (running){
  37.             sourceContext.collect(random.nextInt(100));
  38.         }
  39.         Thread.sleep(2000L);
  40.     }
  41.  
  42.     @Override
  43.     public void cancel() {
  44.         running=false;
  45.     }
  46. }

5.2.7 Flink 支持的数据类型

1. Flink 的类型系统

为了方便地处理数据，Flink 有自己一整套类型系统。Flink 使用“类型信息” （TypeInformation）来统一表示数据类型。TypeInformation 类是 Flink 中所有类型描述符的基类。 它涵盖了类型的一些基本属性，并为每个数据类型生成特定的序列化器、反序列化器和比较器。

2. Flink 支持的数据类型

（1）基本类型

所有 Java 基本类型及其包装类，再加上 Void、String、Date、BigDecimal 和 BigInteger。

（2）数组类型

包括基本类型数组（PRIMITIVE_ARRAY）和对象数组(OBJECT_ARRAY)

（3）复合数据类型

⚫ Java 元组类型（TUPLE）：这是 Flink 内置的元组类型，是 Java API 的一部分。最多

25 个字段，也就是从 Tuple0~Tuple25，不支持空字段

⚫ Scala 样例类及 Scala 元组：不支持空字段

⚫ 行类型（ROW）：可以认为是具有任意个字段的元组,并支持空字段

⚫ POJO：Flink 自定义的类似于 Java bean 模式的类

（4）辅助类型 Option、Either、List、Map 等

（5）泛型类型（GENERIC） Flink 支持所有的 Java 类和 Scala 类。不过如果没有按照上面 POJO 类型的要求来定义， 就会被 Flink 当作泛型类来处理。Flink 会把泛型类型当作黑盒，无法获取它们内部的属性；它 们也不是由 Flink 本身序列化的，而是由 Kryo 序列化的。

在这些类型中，元组类型和 POJO 类型最为灵活，因为它们支持创建复杂类型。而相比之 下，POJO 还支持在键（key）的定义中直接使用字段名，这会让我们的代码可读性大大增加。 所以，在项目实践中，往往会将流处理程序中的元素类型定为 Flink 的 POJO 类型。 Flink 对 POJO 类型的要求如下：

⚫ 类是公共的（public）和独立的（standalone，也就是说没有非静态的内部类）；

⚫ 类有一个公共的无参构造方法；

⚫ 类中的所有字段是 public 且非 final 的；或者有一个公共的 getter 和 setter 方法，这些 方法需要符合 Java bean 的命名规范。

所以我们看到，之前的 UserBehavior，就是我们创建的符合 Flink POJO 定义的数据类型。

3. 类型提示（Type Hints）

Flink 还具有一个类型提取系统，可以分析函数的输入和返回类型，自动获取类型信息， 从而获得对应的序列化器和反序列化器。但是，由于 Java 中泛型擦除的存在，在某些特殊情 况下（比如 Lambda 表达式中），自动提取的信息是不够精细的，这时就需要显式地提供类型信息才能使应用程序正常工作或提高其性能。只有显式地 告诉系统当前的返回类型，才能正确地解析出完整数据。

这是一种比较简单的场景，二元组的两个元素都是基本数据类型。那如果元组中的一个元 素又有泛型，该怎么处理呢？ Flink 专门提供了 TypeHint 类，它可以捕获泛型的类型信息，并且一直记录下来，为运行 时提供足够的信息。我们同样可以通过.returns()方法，明确地指定转换之后的 DataStream 里元 素的类型。

returns(new TypeHint>(){})

5.3 转换算子（Transformation）

数据源读入数据之后，我们就可以使用各种转换算子，将一个或多个 DataStream 转换为 新的DataStream，如图所示。一个 Flink 程序的核心，其实就是所有的转换操作，它们决 定了处理的业务逻辑。

我们可以针对一条流进行转换处理，也可以进行分流、合流等多流转换操作，从而组合成复杂的数据流拓扑。

5.3.1 基本转换算子

1. 映射（map）

map 是大家非常熟悉的大数据操作算子，主要用于将数据流中的数据进行转换，形成新的 数据流。简单来说，就是一个“一一映射”，消费一个元素就产出一个元素

我们只需要基于 DataStrema 调用 map()方法就可以进行转换处理。方法需要传入的参数是 接口 MapFunction 的实现；返回值类型还是 DataStream，不过泛型（流中的元素类型）可能改变。

下面的代码用不同的方式，实现了提取 Event 中的 user 字段的功能。

  1. package com.atguigu.chapter05;
  2.  
  3. import org.apache.flink.api.common.functions.MapFunction;
  4. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  5. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7.  
  8. public class TransFromMapTest {
  9.     public static class Mapper implements MapFunction<Event,String>{
  10.  
  11.         @Override
  12.         public String map(Event event) throws Exception {
  13.             return event.user;
  14.         }
  15.     }
  16.  
  17.     public static void main(String[] args) throws Exception{
  18.         StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
  19.         env.setParallelism(1);
  20.  
  21.         //从元素中读取数据
  22.         DataStreamSource<Event> stream=env.fromElements(new Event("mary","./home",1000L),
  23.         new Event("Bob","./cart",2000L),
  24.         new Event("Alice","./prod?id=100",3000L)
  25.         );
  26.  
  27.         //进行转换计算，提取user字段
  28.  
  29.         //1.使用自定义类，实现MapFunction接口
  30.         SingleOutputStreamOperator<String> result1 = stream.map(new Mapper());
  31.         result1.print();
  32.  
  33.         //2.使用匿名类实现MapFunction接口
  34.         SingleOutputStreamOperator<String> result2 = stream.map(new MapFunction<Event, String>() {
  35.             @Override
  36.             public String map(Event event) throws Exception {
  37.                 return event.user;
  38.             }
  39.         });
  40.         result2.print();
  41.  
  42.         //3.传入Lambda表达式
  43.         SingleOutputStreamOperator<String> result3 = stream.map(data -> data.user);
  44.         result3.print();
  45.  
  46.         env.execute();
  47.  
  48.     }
  49. }

2. 过滤（filter）

filter 转换操作，顾名思义是对数据流执行一个过滤，通过一个布尔条件表达式设置过滤 条件，对于每一个流内元素进行判断，若为 true 则元素正常输出，若为 false 则元素被过滤掉。

 进行 filter 转换之后的新数据流的数据类型与原数据流是相同的。filter 转换需要传入的参数需要实现 FilterFunction 接口，而 FilterFunction 内要实现 filter()方法，就相当于一个返回布尔类型的条件表达式。

代码实现：

  1. package com.atguigu.chapter05;
  2.  
  3. import org.apache.flink.api.common.functions.FilterFunction;
  4. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  5. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7.  
  8. public class TransFromFilterTest {
  9.     public static void main(String[] args) throws Exception {
  10.         StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
  11.         env.setParallelism(1);
  12.  
  13.         //从元素中读取数据
  14.         DataStreamSource<Event> stream=env.fromElements(new Event("mary","./home",1000L),
  15.                 new Event("Bob","./cart",2000L),
  16.                 new Event("Alice","./prod?id=100",3000L)
  17.         );
  18.  
  19.         //1.直接传入一个实现了FilterFunction类的对象
  20.         SingleOutputStreamOperator<Event> result = stream.filter(new MyFilter());
  21.         result.print();
  22.  
  23.         //2.传入一个匿名类实现FilterFunction接口
  24.         SingleOutputStreamOperator<Event> result2 = stream.filter(new FilterFunction<Event>() {
  25.             @Override
  26.             public boolean filter(Event event) throws Exception {
  27.                 if (event.user.equals("mary")) {
  28.                     return true;
  29.                 }
  30.                 return false;
  31.             }
  32.         });
  33.         result2.print();
  34.  
  35.         //传入lambda表达式
  36.         SingleOutputStreamOperator<Event> result3 = stream.filter(data -> data.user.equals("m; ary"));
  37.         result3.print();
  38.  
  39.         env.execute();
  40.     }
  41.  
  42.  
  43.     //实现一个自定义的FilterFunction
  44.     public static class MyFilter implements FilterFunction<Event>{
  45.  
  46.         @Override
  47.         public boolean filter(Event event) throws Exception {
  48.             if (event.user.equals("mary")){
  49.                 return true;
  50.             }
  51.             return false;
  52.         }
  53.     }
  54. }

原文链接：https://blog.csdn.net/JiaXingNashishua/article/details/126692053?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522170392534116800182195992%2522%252C%2522scm%2522%253A%252220140713.130102334.pc%255Fblog.%2522%257D&request_id=170392534116800182195992&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~blog~first_rank_ecpm_v1~rank_v31_ecpm-3-126692053-null-null.nonecase&utm_term=DataStream&spm=1018.2226.3001.4450

标签：DataStream,StreamExecutionEnvironment,Flink,API,flink,env,apache,new
From： https://www.cnblogs.com/sunny3158/p/17936520.html

相关文章

  • Linux驱动开发之Linux内核中的中断处理以及相关API和例程分析
    中断是计算机中实现异步事件处理的一种关键机制。当中断发生时,CPU会暂停当前的任务,转去运行中断服务例程。中断处理完成后,CPU再返回到原来的任务。这使得中断处理具有很高的实时性和响应速度。在Linux内核中,充分利用了中断机制来响应各种硬件和软件事件。在Linux操作系统中，中断......
  • kube-apiserver开启审计
    k8sv1.19.0#/etc/kubernetes/pki/audit-policy.yamlapiVersion:audit.k8s.io/v1beta1kind:Policyrules:-level:Requestresources:-group:""-level:RequestResponseresources:-group:""-level:Metadatares......
  • .Net Core WebAPI 缓存
    Asp.NetCoreWebAPI缓存 一、缓存缓存指在中间层中存储数据的行为，该行为可使后续数据检索更快。从概念上讲，缓存是一种性能优化策略和设计考虑因素。缓存可以显著提高应用性能，方法是提高不常更改（或检索成本高）的数据的就绪性。二、RFC9111在最新的缓存控制规范文件RFC91......
  • 【APIO2016】烟火表演
    先前的题目对slopetrick的认识还不深刻，这题可以看出一个完整的构建过程。题目描述给定一棵有根树，根为\(1\)，边带权，修改边权的代价时修改值与原值差的绝对值，求让所有叶子到根距离相等的最小代价。\(1\leqn\leq3\times10^5,1\leqw\leq10^9\)。算法解析首先有朴......
  • Apipost一键压测参数化功能详解
    最近更新中Apipost对UI页面进行了一些调整，另外一键压测功能支持参数化！本篇文章将详细介绍这些改动！API调试页面的细节改动在请求区填入请求参数或脚本时会有相应的标识如在Query中填入多个参数时上方会展示数量在预、后执行脚本中写入脚本上方会有绿色小点标识一键压测参数化一键压......
  • Apipost一键压测参数化功能详解
    最近更新中Apipost对UI页面进行了一些调整，另外一键压测功能支持参数化！本篇文章将详细介绍这些改动！API调试页面的细节改动在请求区填入请求参数或脚本时会有相应的标识如在Query中填入多个参数时上方会展示数量在预、后执行脚本中写入脚本上方会有绿色小点标识 一键压测参......
  • APISIX 3.0安装及配置使用【转】
    最近一直在研究微服务相关内容，通过对比各大API网关，发现新起之秀APISIX无论从开源程度上来讲还是功能上，都拥有很大的优势。经历了几天折磨一样的学习，目前在本地环境中配置成功了一套，以供自己留存吧，实在是网上的很多文章要么太老了，要么就是乱写一通。 APISIX官方网址：https://ap......
  • Gartner 魔力象限：全生命周期 API 管理 2023 (Magic Quadrant for Full Life Cycle API
    Gartner魔力象限：全生命周期API管理2023GartnerMagicQuadrantforFullLifeCycleAPIManagement2023作者主页：sysin.orgMagicQuadrantforAPIManagementPublished11October2023魔力象限API为数字化转型、现代化和数字化业务生态系统提供了基础，但管理和治理它们具有......
  • Asp.net WebApi Swagger Tag 标记分组归纳显示Api接口路由
    官方文档说明地址https://swagger.io/docs/specification/2-0/grouping-operations-with-tags/创建一个自定义的特性类publicclassControllerGroupAttribute:Attribute{publicControllerGroupAttribute(stringgroupName){if......
  • Amazon API Gateway CORS 实战
    AmazonAPIGateway是构建和部署RESTfulAPI的托管服务，而跨域资源共享（CORS）是为了在Web应用中安全实现跨域数据传输而设计的标准。在本文中，我们将深入探讨如何在AmazonAPIGateway上实战配置CORS，确保安全地处理跨域请求。1.CORS概述1.1同源策略回顾同源策略是浏览器的安......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99