使用 Apache Flink 将数据推送到 S3答案

【问题标题】：Pushing data to S3 with Apache Flink使用 Apache Flink 将数据推送到 S3
【发布时间】：2017-01-09 20:42:08
【问题描述】：

我有一个将数据推送到 S3 存储桶的小型测试项目。但是，看起来我没有读取 core-site.xml 文件，因为我收到了错误java.io.IOException: No file system found with scheme s3a。如何正确读取 core-site.xml 文件并将数据推送到 S3？

这是代码：

public class S3Sink {
public static void main(String[] args) throws Exception {
    Map<String, String> configs = ConfigUtils.loadConfigs(“path/to/config.yaml");

    final ParameterTool parameterTool = ParameterTool.fromMap(configs);

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.getConfig().disableSysoutLogging();
    env.getConfig().setGlobalJobParameters(parameterTool); 

    DataStream<String> messageStream = env
            .addSource(new FlinkKafkaConsumer09<String>(
                    parameterTool.getRequired("kafka.topic"),
                    new SimpleStringSchema(),
                    parameterTool.getProperties()));

    String id = UUID.randomUUID().toString();
    messageStream.writeAsText("s3a://flink-test/" + id + ".txt").setParallelism(1);

    env.execute();
}

这是 flink-conf.yaml 文件中的配置更改以引用 core-site.xml 文件：

fs.hdfs.hadoopconf: /path/to/core-site/etc/hadoop/

这是我的 core-site.xml：

<configuration>
<property>
    <name>fs.defaultFS</name>
    <value>hdfs://localhost:9000</value>
</property>
<property>
    <name>fs.s3.impl</name>
    <value>org.apache.hadoop.fs.s3a.S3AFileSystem</value>
</property>

<!-- Comma separated list of local directories used to buffer
     large results prior to transmitting them to S3. -->
<property>
    <name>fs.s3a.buffer.dir</name>
    <value>/tmp</value>
</property>

<!-- set your AWS ID using key defined in org.apache.hadoop.fs.s3a.Constants -->
<property>
    <name>fs.s3a.awsAccessKeyId</name>
    <value>*****</value>
</property>
<!-- set your AWS access key -->
<property>
    <name>fs.s3a.awsSecretAccessKey</name>
    <value>*****</value>
</property>

【问题讨论】：

如果将fs.hdfs.hadoopconf 设置为包含core-site.xml 的文件夹，它会起作用吗？还要确保$HADOOP_HOME 环境变量设置正确。
我正在使用 IntelliJ 并将环境变量 HADOOP_HOME 设置为 core-site.xml 路径。我在本地运行程序，所以 fs.hdfs.hadoopconf 设置没有帮助。

标签： xml hadoop amazon-s3 apache-flink

【解决方案1】：

没有读入 core-site.xml 文件的原因是 Hadoop 的文件结构。我有HADOOP_HOME=path/to/dir/etc/hadoop。然而，Hadoop 在其文件结构中查找 etc/hadoop 以查找 core-site.xml。要在 HADOOP_HOME 环境变量中正确读取路径，应将其列为HADOOP_HOME=path/to/dir。

另一个问题是为什么数据没有推送到 S3。这是因为我使用的是流处理。批处理可以将数据推送到 S3，但流处理不是因为 S3 如何将数据存储为键/值存储，并且不能追加新数据，只能替换。对于流处理，Flink 不断地将数据附加到 S3 不允许的同一文件中，因此没有数据被推送到 S3。所以这段代码适用于将批次推送到 S3

    ExecutionEnvironment ee = ExecutionEnvironment.getExecutionEnvironment();
    DataSet dataSet = ee.readTextFile("/Users/name/Desktop/flinkoutputtest.txt");
    dataSet.writeAsText("s3://flink-test/flink-output/testdoc.txt").setParallelism(1);
    ee.execute();

【讨论】：