Commit 79b5d446 by 余根龙

增加jobname

parent ee2b1e28
......@@ -5,12 +5,8 @@ import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Stage;
import com.zhht.irn.sink.GreenRadioSink;
import com.zhht.irn.utils.FlinkUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.net.URL;
import java.util.List;
/**
......@@ -25,7 +21,6 @@ public class GreenRadioJob {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
ParameterTool kafkaTool = ParameterTool.fromArgs(args);
String topic = kafkaTool.get("kafka.input.topics");
System.out.println("***topic:"+topic);
FlinkUtils.createKafkaStream(args, env,topic,"GreenRadioJob")
.map(record -> {
// 字符串解析为实体类
......@@ -71,6 +66,6 @@ public class GreenRadioJob {
return cycle;
})
.addSink(new GreenRadioSink());
env.execute();
env.execute("GreenRadioJob");
}
}
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment