Hi guys, I am using flink version 1.7.2. I am trying to write to hdfs sink from my flink job. I setup HADOOP_HOME. Here is the debug log for this :
2020-03-19 18:59:34,316 DEBUG org.apache.flink.runtime.util.HadoopUtils - Cannot find hdfs-default configuration-file path in Flink config. 2020-03-19 18:59:34,317 DEBUG org.apache.flink.runtime.util.HadoopUtils - Cannot find hdfs-site configuration-file path in Flink config. 2020-03-19 18:59:34,317 DEBUG org.apache.flink.runtime.util.HadoopUtils - Adding /home/was/HDFSConf/conf/core-site.xml to hadoop configuration 2020-03-19 18:59:34,317 DEBUG org.apache.flink.runtime.util.HadoopUtils - Adding /home/was/HDFSConf/conf/hdfs-site.xml to hadoop configuration 2020-03-19 18:59:34,344 INFO org.apache.flink.runtime.security.modules.HadoopModule - Hadoop user set to kafka (auth:KERBEROS) This is what my streaming file sink code looks like. val sink: StreamingFileSink[String] = StreamingFileSink .forRowFormat(new Path("hdfs://tmp/auditlog/"), new SimpleStringEncoder[String]("UTF-8")) .withRollingPolicy(DefaultRollingPolicy.create() .withRolloverInterval(TimeUnit.MINUTES.toMillis(15)) .withInactivityInterval(TimeUnit.MINUTES.toMillis(5)) .withMaxPartSize(1024 * 1024 * 1024) .build()) .build() result.addSink(sink).name("HDFSSink") When I run the job I get this error stack trace : INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Sink: HDFSSink (1/1) (27b62d6294da47491041d750daf421a0) switched from RUNNING to FAILED. java.io.IOException: Cannot instantiate file system for URI: hdfs://tmp/auditlog at org.apache.flink.runtime.fs.hdfs.HadoopFsFactory.create(HadoopFsFactory.java:187) at org.apache.flink.core.fs.FileSystem.getUnguardedFileSystem(FileSystem.java:399) at org.apache.flink.core.fs.FileSystem.get(FileSystem.java:318) at org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.<init>(Buckets.java:112) at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink$RowFormatBuilder.createBuckets(StreamingFileSink.java:242) at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.initializeState(StreamingFileSink.java:327) at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.tryRestoreFunction(StreamingFunctionUtils.java:178) at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.restoreFunctionState(StreamingFunctionUtils.java:160) at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.initializeState(AbstractUdfStreamOperator.java:96) at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:278) at org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState(StreamTask.java:738) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:289) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:704) at java.lang.Thread.run(Thread.java:748) Caused by: java.lang.IllegalArgumentException: java.net.UnknownHostException: tmp at org.apache.hadoop.security.SecurityUtil.buildTokenService(SecurityUtil.java:378) at org.apache.hadoop.hdfs.NameNodeProxies.createNonHAProxy(NameNodeProxies.java:320) at org.apache.hadoop.hdfs.NameNodeProxies.createProxy(NameNodeProxies.java:176) at org.apache.hadoop.hdfs.DFSClient.<init>(DFSClient.java:687) at org.apache.hadoop.hdfs.DFSClient.<init>(DFSClient.java:628) at org.apache.hadoop.hdfs.DistributedFileSystem.initialize(DistributedFileSystem.java:149) Why is it trying to connect to /tmp ? Is it not supposed to get the namenodes from the core-site.xml and hdfs-site.xml ? Can you please help with the correct way to configure hdfs sink. Best, Nick.