随机产生数据然后将产生的数据写入到hdfs 中。
2. 随机数据源代码:
package com.wudl.flink.hdfs.source; import org.apache.flink.api.java.tuple.Tuple; import org.apache.flink.api.java.tuple.Tuple4; import org.apache.flink.streaming.api.functions.source.SourceFunction; import java.text.SimpleDateFormat; import java.util.Date; import java.util.Random; public class MySource implements SourceFunction3. hdfssink{ private boolean isRunning = true; String[] citys = {"北京","广东","山东","江苏","河南","上海","河北","浙江","香港","山西","陕西","湖南","重庆","福建","天津","云南","四川","广西","安徽","海南","江西","湖北","山西","辽宁","内蒙古"}; int i = 0; @Override public void run(SourceContext ctx) throws Exception { Random random = new Random(); SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); while (isRunning) { int number = random.nextInt(4) + 1; Integer id = i++; String eventTime = df.format(new Date()); String address = citys[random.nextInt(citys.length)]; int productId = random.nextInt(25); ctx.collect(id+","+eventTime +","+ address +","+ productId); Thread.sleep(500); } } @Override public void cancel() { isRunning = false; } }
需要注意的怎么设置文件的前缀和后缀以及 文件的大小 。
package com.wudl.flink.hdfs.sink; import org.apache.flink.api.common.serialization.SimpleStringEncoder; import org.apache.flink.api.connector.sink.Sink; import org.apache.flink.api.java.tuple.Tuple4; import org.apache.flink.connector.file.sink.FileSink; import org.apache.flink.core.fs.Path; import org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig; import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.DateTimeBucketAssigner; import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy; import java.util.concurrent.TimeUnit; public class HdfsSink { public static FileSink4.主类getHdfsSink() { OutputFileConfig config = OutputFileConfig .builder() .withPartPrefix("wudl") .withPartSuffix(".txt") .build(); FileSink finkSink = FileSink.forRowFormat(new Path("hdfs://192.168.1.161:8020/FlinkFileSink/wudlfile"), new SimpleStringEncoder ("UTF-8")) .withRollingPolicy(DefaultRollingPolicy.builder() 每隔15分钟生成一个新文件 .withRolloverInterval(TimeUnit.MINUTES.toMinutes(1)) //每隔5分钟没有新数据到来,也把之前的生成一个新文件 .withInactivityInterval(TimeUnit.MINUTES.toMinutes(5)) .withMaxPartSize(1024 * 1024 * 1024) .build()) .withOutputFileConfig(config) .withBucketAssigner(new DateTimeBucketAssigner("yyyy-MM-dd")) .build(); return finkSink; } }
package com.wudl.flink.hdfs; import com.wudl.flink.hdfs.sink.HdfsSink; import com.wudl.flink.hdfs.source.MySource; import org.apache.commons.lang3.SystemUtils; import org.apache.flink.api.common.restartstrategy.RestartStrategies; import org.apache.flink.api.common.time.Time; import org.apache.flink.api.java.tuple.Tuple4; import org.apache.flink.connector.file.sink.FileSink; import org.apache.flink.runtime.state.filesystem.FsStateBackend; import org.apache.flink.streaming.api.CheckpointingMode; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import java.util.concurrent.TimeUnit; public class Application { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); env.getCheckpointConfig().setCheckpointStorage("hdfs://192.168.1.161:8020/flink-checkpoint/checkpoint"); //设置两个Checkpoint 之间最少等待时间,如设置Checkpoint之间最少是要等 500ms(为了避免每隔1000ms做一次Checkpoint的时候,前一次太慢和后一次重叠到一起去了 --//默认是0 env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500); //设置Checkpoint的时间间隔为1000ms做一次Checkpoint/其实就是每隔1000ms发一次Barrier! env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); //设置checkpoint的超时时间,如果 Checkpoint在 60s内尚未完成说明该次Checkpoint失败,则丢弃。 env.getCheckpointConfig().setCheckpointTimeout(10000L); //设置同一时间有多少个checkpoint可以同时执行 env.getCheckpointConfig().setMaxConcurrentCheckpoints(2); // 设置重启策略 // 一个时间段内的最大失败次数 env.setRestartStrategy(RestartStrategies.failureRateRestart(3, // 衡量失败次数的是时间段 Time.of(5, TimeUnit.MINUTES), // 间隔 Time.of(10, TimeUnit.SECONDS) )); DataStreamSource结果mySouceData = env.addSource(new MySource()); FileSink hdfsSink = HdfsSink.getHdfsSink(); mySouceData.sinkTo(hdfsSink ); mySouceData.print(); env.execute(); } }
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)