我正在flink cluster上部署作业。s3路径类似于:
网址:s3a://bucket name/pre/pre1/original/2019\u 12\u 19/file.parquet.gz
我把它读作:
val job = Job.getInstance
FileInputFormat.addInputPath(
job,
new org.apache.hadoop.fs.Path(url)
)
val hadoopInputFormat =
new HadoopInputFormat(
new AvroParquetInputFormat[GenericRecord],
classOf[Void],
classOf[GenericRecord],
job
)
val data: List[tuple.Tuple2[Void, GenericRecord]] =
env.createInput(hadoopInputFormat).collect().asScala.toList
虽然作业运行的是带有sbt run的文件,但它没有提交。
此外,当部署作业且s3 url的类型为:
网址:s3a://bucket name/2019\u 12\u 19/file.parquet.gz
提交成功。
val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
env.setParallelism(parallelism)
依赖项:
"org.apache.flink" %% "flink-scala" % "1.10.0" % "provided"
"org.apache.flink" % "flink-s3-fs-hadoop" % "1.10.0"
"org.apache.flink" %% "flink-hadoop-compatibility" % "1.10.0"
"org.apache.hadoop" % "hadoop-mapreduce-client-core" % ""3.1.1""
"org.apache.hadoop" % "hadoop-aws" % "2.7.2"
"org.apache.httpcomponents" % "httpcore" % "4.2.5"
"org.apache.httpcomponents" % "httpclient" % "4.2.5"
"org.apache.flink" %% "flink-streaming-scala" % "1.10.0" % "provided"
Exception:
Caused by: java.io.IOException: Class class com.amazonaws.auth.InstanceProfileCredentialsProvider does not implement AWSCredentialsProvider
at org.apache.hadoop.fs.s3a.S3AUtils.createAWSCredentialProvider(S3AUtils.java:623)
at org.apache.hadoop.fs.s3a.S3AUtils.createAWSCredentialProviderSet(S3AUtils.java:566)
at org.apache.hadoop.fs.s3a.DefaultS3ClientFactory.createS3Client(DefaultS3ClientFactory.java:52)
at org.apache.hadoop.fs.s3a.S3AFileSystem.initialize(S3AFileSystem.java:256)
at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:3354)
at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:124)
at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:3403)
at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:3371)
at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:477)
at org.apache.hadoop.fs.Path.getFileSystem(Path.java:361)
at org.apache.hadoop.mapreduce.lib.input.FileInputFormat.singleThreadedListStatus(FileInputFormat.java:302)
at org.apache.hadoop.mapreduce.lib.input.FileInputFormat.listStatus(FileInputFormat.java:274)
at org.apache.parquet.hadoop.ParquetInputFormat.listStatus(ParquetInputFormat.java:349)
at org.apache.hadoop.mapreduce.lib.input.FileInputFormat.getSplits(FileInputFormat.java:396)
at org.apache.parquet.hadoop.ParquetInputFormat.getSplits(ParquetInputFormat.java:304)
at org.apache.flink.api.java.hadoop.mapreduce.HadoopInputFormatBase.createInputSplits(HadoopInputFormatBase.java:159)
at org.apache.flink.api.java.hadoop.mapreduce.HadoopInputFormatBase.createInputSplits(HadoopInputFormatBase.java:59)
at org.apache.flink.runtime.executiongraph.ExecutionJobVertex.<init>(ExecutionJobVertex.java:257)
core-site.xml:
<configuration>
<property>
<name>fs.s3a.impl</name>
<value>org.apache.hadoop.fs.s3a.S3AFileSystem</value>
</property>
<property>
<name>fs.s3a.buffer.dir</name>
<value>/tmp</value>
</property>
<property>
<name>fs.s3a.access.key</name>
<value>iam</value>
</property>
<property>
<name>fs.s3a.secret.key</name>
<value>iam</value>
</property>
<property>
<name>fs.s3a.endpoint</name>
<value>https://s3.us-east-1.amazonaws.com</value>
</property>
<property>
<name>fs.s3a.path.style.access</name>
<value>true</value>
</property>
<property>
<name>fs.s3a.aws.credentials.provider</name>
<value>
com.amazonaws.auth.InstanceProfileCredentialsProvider,
org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider,
com.amazonaws.auth.EnvironmentVariableCredentialsProvider
</value>
</property>
</configuration>
1条答案
按热度按时间uelo1irk1#
从Flink1.10开始,只能使用
flink-s3-fs-hadoop
作为插件。您基本上需要将jar添加到您的flink dist中,如下所示。
把它从你的肥jar里拿出来。