??????flink ????hadoop????????

Fri, 29 May 2020 00:44:49 -0700 

??????????????????OK??




------------------ ???????? ------------------
??????:&nbsp;"????"<[email protected]&gt;;
????????:&nbsp;2020??5??29??(??????) ????3:20
??????:&nbsp;"user-zh"<[email protected]&gt;;

????:&nbsp;Re:flink ????hadoop????????









???????????????????????????? ?????????????????????????? ??????hdfs-site.xml 
??core-site.xml ??????????????????????











?? 2020-05-29 15:06:21??"??????????????" <[email protected]&gt; ??????
&gt;???????????????? ??
&gt;hadoop????TestHACluster??????????api????????????????path 
hdfs://TestHACluster/user/flink/test&amp;nbsp;&amp;nbsp;
&gt;??????????TestHACluster:8020?? 
????????????????????????????????????????????????
&gt;??????????????????????????hive????????????????????????????TestHACluster:8020
&gt;StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
&gt;DataStream<String&amp;gt; input = 
env.fromCollection(Arrays.asList("1","2"));
&gt;
&gt;StreamingFileSink<String&amp;gt; sink = StreamingFileSink.forRowFormat(
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new 
Path("hdfs://TestHACluster/user/flink/test"),
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; new 
SimpleStringEncoder<String&amp;gt;(Encoder.UTF_8))
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; .withBucketAssigner(new 
DateTimeBucketAssigner<&amp;gt;())
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; .build();
&gt;input.addSink(sink);
&gt;env.execute();

回复