public static void main(String [] args) throws Exception {
//构建执行任务环境以及任务的启动的入口, 存储全局相关的参数
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//相同类型元素的数据流 source
DataStream<String> stringDS1 = env.fromElements("java,SpringBoot", "spring cloud,redis", "kafka,小滴课堂");
stringDS1.print("stringDS1");
DataStream<String> stringDS2 = env.fromCollection(Arrays.asList("微服务项目大课,java","alibabacloud,rabbitmq","hadoop,hbase"));
stringDS2.print("stringDS2");
DataStreamSource<Long> longDS3 = env.fromSequence(0,10);
longDS3.print("longDS3");
//DataStream需要调用execute,可以取个名称
env.execute("xdclass job");
}
public static void main(String [] args) throws Exception {
//构建执行任务环境以及任务的启动的入口, 存储全局相关的参数
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStream<String> textDS = env.readTextFile("/Users/xdclass/Desktop/xdclass_access.log");
//DataStream<String> textDS = env.readTextFile("hdfs://xdclass_node:8010/file/log/words.txt");
textDS.print();
env.execute("xdclass job");
}
public static void main(String [] args) throws Exception {
//构建执行任务环境以及任务的启动的入口, 存储全局相关的参数
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStream<String> stringDataStream = env.socketTextStream("127.0.0.1",8888);
stringDataStream.print();
env.execute(" job");
}
@Data
@AllArgsConstructor
@NoArgsConstructor
public class VideoOrder {
private String tradeNo;
private String title;
private int money;
private int userId;
private Date createTime;
}
public class VideoOrderSource extends RichParallelSourceFunction<VideoOrder> {
private volatile Boolean flag = true;
private Random random = new Random();
private static List<String> list = new ArrayList<>();
static {
list.add("spring boot2.x课程");
list.add("微服务SpringCloud课程");
list.add("RabbitMQ消息队列");
list.add("Kafka课程");
list.add("Flink流式技术课程");
list.add("工业级微服务项目大课训练营");
list.add("Linux课程");
}
@Override
public void run(SourceContext<VideoOrder> ctx) throws Exception {
while (flag){
Thread.sleep(1000);
String id = UUID.randomUUID().toString();
int userId = random.nextInt(10);
int money = random.nextInt(100);
int videoNum = random.nextInt(list.size());
String title = list.get(videoNum);
ctx.collect(new VideoOrder(id,title,money,userId,new Date()));
}
}
/**
* 取消任务
*/
@Override
public void cancel() {
flag = false;
}
}
public static void main(String [] args) throws Exception {
//构建执行任务环境以及任务的启动的入口, 存储全局相关的参数
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<VideoOrder> videoOrderDataStream = env.addSource(new VideoOrderSource());
videoOrderDataStream.print();
//DataStream需要调用execute,可以取个名称
env.execute("custom source job");
}
不断产生很多订单