Flink

做为一款流式计算框架#xff0c;它可用来做批处理#xff0c;也可以用来做流处理#xff0c;这个

Data

flink在批/流处理中常见的source主要有两大类…Data

Sources

基于本地集合的sourceCollection-based-source

自定义Source

使用env.fromElements()这种方式也支持Tuple自定义对象等复合形式。

注意类型要一致不一致可以用Object接收但是使用会报错比如env.fromElements(haha,

1);

|使用env.fromCollection(),这种方式支持多种Collection的具体类型如ListSetQueue

使用env.generateSequence()方法创建基于Sequence的DataStream

使用env.fromSequence()方法创建基于开始和结束的DataStream

一般用于学习测试时编造数据时使用

org.apache.flink.api.java.tuple.Tuple2;

import

org.apache.flink.streaming.api.datastream.DataStreamSource;

import

org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import

java.util.ArrayList;

StreamExecutionEnvironment.getExecutionEnvironment();//

各种获取数据的SourceDataStreamSourceString

dataStreamSource

kongniqiwa);dataStreamSource.print();//

演示一个错误的//DataStreamSourceObject

dataStreamSource2

1,3.0f);//dataStreamSource2.print();DataStreamSourceTuple2String,

Integer

{hello,world};System.out.println(arr.length);System.out.println(Arrays.toString(arr));ListString

list

Arrays.asList(arr);System.out.println(list);env.fromElements(Arrays.asList(arr),Arrays.asList(arr),Arrays.asList(arr)).print();//

Collection

ArrayList();list1.add(python);list1.add(scala);list1.add(java);DataStreamSourceString

ds1

env.fromCollection(list1);DataStreamSourceString

ds2

env.fromCollection(Arrays.asList(arr));//

ds3

下面的代码不运行所以这句话要放在最后。

env.execute(获取预定义的Source);}

package

org.apache.flink.api.java.tuple.Tuple2;

import

org.apache.flink.streaming.api.datastream.DataStreamSource;

import

org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import

java.io.File;

StreamExecutionEnvironment.getExecutionEnvironment();//

获取并行度System.out.println(env.getParallelism());//

讲第二种Source

File(./);System.out.println(file.getAbsoluteFile());System.out.println(file2.getAbsoluteFile());DataStreamSourceString

ds1

env.readTextFile(datas/wc.txt);ds1.print();//

还可以获取hdfs路径上的数据DataStreamSourceString

ds2

env.readTextFile(hdfs://bigdata01:9820/home/a.txt);ds2.print();//

execute

下面的代码不运行所以这句话要放在最后。

env.execute(获取预定义的Source);}

Socket

方法是一个非并行的Source该方法需要传入两个参数第一个是指定的IP地址或主机名第二个是端口号即从指定的Socket读取数据创建DataStream。

该方法还有多个重载的方法其中一个是socketTextStream(String

hostname,

maxRetry)这个重载的方法可以指定行分隔符和最大重新连接次数。

这两个参数默认行分隔符是”\n”最大重新连接次数为0。

如果使用socketTextStream读取数据在启动Flink程序之前必须先启动一个Socket服务为了方便Mac或Linux用户可以在命令行终端输入nc

-lk

8888启动一个Socket服务并在命令行中向该Socket服务发送数据。

Windows用户可以在百度中搜索windows安装netcat命令。

使用nc

--向8888端口发送消息这个命令先运行如果先运行java程序会报错

-lp

//socketTextStream创建的DataStream不论怎样并行度永远是1

public

{//local模式默认的并行度是当前机器的逻辑核的数量StreamExecutionEnvironment

env

StreamExecutionEnvironment.getExecutionEnvironment();int

parallelism0

env.getParallelism();System.out.println(执行环境默认的并行度

parallelism0);DataStreamSourceString

lines

env.socketTextStream(localhost,

parallelism

lines.getParallelism();System.out.println(SocketSource的并行度

parallelism);SingleOutputStreamOperatorString

words

{collector.collect(word);}}});int

parallelism2

words.getParallelism();System.out.println(调用完FlatMap后DataStream的并行度

parallelism2);words.print();env.execute();}

import

org.apache.flink.api.common.RuntimeExecutionMode;

import

org.apache.flink.api.common.functions.FlatMapFunction;

import

org.apache.flink.api.common.functions.MapFunction;

import

org.apache.flink.api.java.functions.KeySelector;

import

org.apache.flink.api.java.tuple.Tuple2;

import

org.apache.flink.streaming.api.datastream.DataStream;

import

org.apache.flink.streaming.api.datastream.KeyedStream;

import

org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;

import

org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import

org.apache.flink.util.Collector;public

class

1.env-准备环境StreamExecutionEnvironment

env

StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//TODO

socketDS

env.socketTextStream(bigdata01,

8889);//TODO

3.transformation-数据转换处理//3.1对每一行数据进行分割并压扁DataStreamString

wordsDS

{out.collect(word);}}});//3.2每个单词记为单词,1DataStreamTuple2String,

Integer

1);}});//3.3分组KeyedStreamTuple2String,

Integer,

value.f0;}});//3.4聚合SingleOutputStreamOperatorTuple2String,

Integer

4.sink-数据输出result.print();//TODO

自定义数据源

RichSourceFunction:多功能非并行数据源(并行度只能1)

--类

ParallelSourceFunction:并行数据源(并行度能够1)

--接口

RichParallelSourceFunction:多功能并行数据源(并行度能够1)

--类

org.apache.flink.streaming.api.datastream.DataStreamSource;

import

org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import

org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;

import

org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import

org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import

org.apache.flink.streaming.api.functions.source.SourceFunction;import

java.util.Random;

每隔1秒随机生成一条订单信息(订单ID、用户ID、订单金额、时间戳)*

要求:*

RichParallelSourceFunctionOrderInfo

class

OrderInfo();orderInfo.setOrderId(UUID.randomUUID().toString());orderInfo.setUid(random.nextInt(3));orderInfo.setMoney(random.nextInt(101));orderInfo.setTimeStamp(System.currentTimeMillis());ctx.collect(orderInfo);Thread.sleep(1000);//

间隔1s}}//

StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);//

将自定义的数据源放入到env中DataStreamSource

dataStreamSource

MySource())/*.setParallelism(1)*/;System.out.println(dataStreamSource.getParallelism());dataStreamSource.print();env.execute();}}

通过ParallelSourceFunction创建可并行Source

/***

StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSourceString

mySource

MySource()).setParallelism(6);mySource.print();env.execute();}public

static

{ctx.collect(UUID.randomUUID().toString());/*如果不设置无限循环可以看出设置了多少并行度就打印出多少条数据*/}Overridepublic

void

如果代码换成ParallelSourceFunction每次生成12个数据假如是12核数的话。

总结Rich富函数总结

open方法实例化的时候会执行一次多个并行度会执行多次的哦因为是多个实例了

close方法销毁实例的时候会执行一次多个并行度会执行多次的哦

getRuntimeContext

https://nightlies.apache.org/flink/flink-docs-release-1.13/zh/docs/connectors/datastream/kafka/

dependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-kafka_2.11/artifactIdversion${flink.version}/version

/dependency

/opt/installs/kafka3/bin/kafka-topics.sh

--bootstrap-server

org.apache.flink.api.common.functions.FilterFunction;

import

org.apache.flink.api.common.serialization.SimpleStringSchema;

import

org.apache.flink.streaming.api.datastream.DataStreamSource;

import

org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import

org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;import

class

StreamExecutionEnvironment.getExecutionEnvironment();Properties

properties

Properties();properties.setProperty(bootstrap.servers,

bigdata01:9092);properties.setProperty(group.id,

kafkaSource

FlinkKafkaConsumerString(topic1,new

SimpleStringSchema(),properties);DataStreamSourceString

dataStreamSource

以下代码跟flink消费kakfa数据没关系仅仅是将需求搞的复杂一点而已//

返回true

直接丢弃dataStreamSource.filter(new

{Overridepublic

word.contains(success);}}).print();env.execute();}