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

Fri, 29 May 2020 00:36:50 -0700 

thanks very much ??????
????resource???? ????hdfs-site.xml




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

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



&nbsp;&nbsp; ????cdh??????hdfs????????????8020?????????? 
hdfs://TestHACluster/user/flink/test ????&nbsp; 
hdfs://TestHACluster:8020/user/flink/test????????????????????
&nbsp;&nbsp; 
????????????flink????????TestHACluster????Namespace??hdfs????HA??????????????????????hive-site.xml??hdfs-site.xml????????????????


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

回复