请确认这是使用Flink将数据流传输到Hadoop的正确方式

我需要一些关于Flink Streaming的帮助。我在下面生成了一个简单的Hello世界类型的代码。这将从RabbitMQ流式传输Avro消息，并将其持久化到HDFS。我希望有人能复习一下代码，也许它能帮助其他人。

我发现的大多数Flink流式传输的例子都会将结果发送到std-out。实际上我想把数据保存到Hadoop中。我读到，理论上，你可以和Flink一起流到任何你喜欢的地方。实际上，我还没有发现任何将数据保存到HDFS的例子。但是，根据我发现的例子，以及试验和错误，我得到了以下代码。

这里的数据源是RabbitMQ。我使用客户端应用程序将"MyAvroObjects"发送到RabbitMQ。MyAvroObject.java（不包括在内）是从avro IDL生成的。。。可以是任何avro消息。

下面的代码使用RabbitMQ消息，并将其保存到HDFS中，作为avro文件。。。这就是我所希望的。

package com.johanw.flink.stackoverflow;
import java.io.IOException;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.mapred.AvroKey;
import org.apache.avro.mapred.AvroOutputFormat;
import org.apache.avro.mapred.AvroWrapper;
import org.apache.avro.mapreduce.AvroJob;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.hadoop.mapred.HadoopOutputFormat;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.FileSinkFunctionByMillis;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.util.serialization.DeserializationSchema;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class RMQToHadoop {
    public class MyDeserializationSchema implements DeserializationSchema<MyAvroObject> {
        private static final long serialVersionUID = 1L;
        @Override
        public TypeInformation<MyAvroObject> getProducedType() {
             return TypeExtractor.getForClass(MyAvroObject.class);
        }
        @Override
        public MyAvroObject deserialize(byte[] array) throws IOException {
            SpecificDatumReader<MyAvroObject> reader = new SpecificDatumReader<MyAvroObject>(MyAvroObject.getClassSchema());
            Decoder decoder = DecoderFactory.get().binaryDecoder(array, null);
            MyAvroObject MyAvroObject = reader.read(null, decoder);
            return MyAvroObject;
        }
        @Override
        public boolean isEndOfStream(MyAvroObject arg0) {
            return false;
        }
    }
    private String hostName;
    private String queueName;
    public final static String path = "/hdfsroot";
    private static Logger logger = LoggerFactory.getLogger(RMQToHadoop.class);
    public RMQToHadoop(String hostName, String queueName) {
        super();
        this.hostName = hostName;
        this.queueName = queueName;
    }
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    public void run() {
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        logger.info("Running " + RMQToHadoop.class.getName());
        DataStream<MyAvroObject> socketStockStream = env.addSource(new RMQSource<>(hostName, queueName, new MyDeserializationSchema()));
        Job job;
        try {
            job = Job.getInstance();
            AvroJob.setInputKeySchema(job, MyAvroObject.getClassSchema());
        } catch (IOException e1) {
            e1.printStackTrace();
        }
        try {
            JobConf jobConf = new JobConf(Job.getInstance().getConfiguration());
            jobConf.set("avro.output.schema", MyAvroObject.getClassSchema().toString());
            org.apache.avro.mapred.AvroOutputFormat<MyAvroObject> akof = new AvroOutputFormat<MyAvroObject>();
            HadoopOutputFormat<AvroWrapper<MyAvroObject>, NullWritable> hof = new HadoopOutputFormat<AvroWrapper<MyAvroObject>, NullWritable>(akof, jobConf);
            FileSinkFunctionByMillis<Tuple2<AvroWrapper<MyAvroObject>, NullWritable>> fileSinkFunctionByMillis = new FileSinkFunctionByMillis<Tuple2<AvroWrapper<MyAvroObject>, NullWritable>>(hof, 10000l);
            org.apache.hadoop.mapred.FileOutputFormat.setOutputPath(jobConf, new Path(path));
            socketStockStream.map(new MapFunction<MyAvroObject, Tuple2<AvroWrapper<MyAvroObject>, NullWritable>>() {
                private static final long serialVersionUID = 1L;
                @Override
                public Tuple2<AvroWrapper<MyAvroObject>, NullWritable> map(MyAvroObject envelope) throws Exception {
                    logger.info("map");
                    AvroKey<MyAvroObject> key = new AvroKey<MyAvroObject>(envelope);
                    Tuple2<AvroWrapper<MyAvroObject>, NullWritable> tupple = new Tuple2<AvroWrapper<MyAvroObject>, NullWritable>(key, NullWritable.get());
                    return tupple;
                }
            }).addSink(fileSinkFunctionByMillis);
            try {
                env.execute();
            } catch (Exception e) {
                logger.error("Error while running " + RMQToHadoop.class + ".", e);
            }
        } catch (IOException e) {
            logger.error("Error while running " + RMQToHadoop.class + ".", e);
        }
    }
    public static void main(String[] args) throws IOException {
        RMQToHadoop toHadoop = new RMQToHadoop("localhost", "rabbitTestQueue");
        toHadoop.run();
    }
}

如果您更喜欢RabbitMQ以外的其他源，那么使用其他源可以很好地工作。例如，使用Kafka消费者：

import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer082;
...
DataStreamSource<MyAvroObject> socketStockStream = env.addSource(new FlinkKafkaConsumer082<MyAvroObject>(topic, new MyDeserializationSchema(), sourceProperties));

问题：

请复习。这是将数据保存到HDFS的良好做法吗？
如果流媒体的过程导致了一个问题，比如说在连载过程中。它生成和异常，然后代码就退出了。Spark流依赖于Yarn自动重启应用程序。这也是使用Flink时的好做法吗？
我正在使用FileSinkFunctionByMillis。实际上，我希望使用类似HdfsSinkFunction的东西，但这并不存在。所以FileSinkFunctionByMillis是最接近这一点的，这对我来说很有意义。同样，我发现的文档没有任何解释，所以我只是猜测。
当我在本地运行它时，我会发现一个目录结构，比如"C:\hdfsroot_temporary\0_temporary \temppt__0000_r_000001_0"，它是…basare。有什么想法吗？

顺便说一句，当你想把数据保存回Kafka时，我可以用。。。

Properties destProperties = new Properties();
destProperties.setProperty("bootstrap.servers", bootstrapServers);
FlinkKafkaProducer<MyAvroObject> kafkaProducer = new FlinkKafkaProducer<L3Result>("MyKafkaTopic", new MySerializationSchema(), destProperties);

非常感谢！！！！

我认为可以使用FileSinkFunctionByMillis，但这意味着流媒体程序是不容错的。这意味着，如果你的源代码、机器或写入失败，那么你的程序将崩溃，无法恢复。

我建议您考虑使用RollingSink(https://ci.apache.org/projects/flink/flink-docs-release-0.10/apis/streaming_guide.html#hadoop-文件系统）。这可以用来创建类似Flum的管道，将数据摄入HDFS（或其他文件系统）。滚动接收器是一个可恢复的接收器，这意味着你的程序是容错的，因为Kafka消费者也是容错的。此外，您还可以指定一个自定义Writer来以您想要的任何格式写入数据，例如Avro。

相关内容

最新更新

热门标签：