使用自定义AvroParquetWriter的Flink BucketingSink创建空文件

问题描述

我为BucketingSink创建了一个作家.接收器和写入器可以正常工作，但是当写入器将avro通用记录写入镶木地板时，该文件是从进行中创建的，尚待完成.但是文件为0字节，为空.谁能告诉我代码有什么问题吗?我试图将AvroParquetWriter的初始化放在open()方法上，但结果仍然相同.

I have created a writer for BucketingSink. The sink and writer works without error but when it comes to the writer writing avro genericrecord to parquet, the file was created from in-progress, pending to complete. But the files are empty with 0 bytes. Can anyone tell me what is wrong with the code ? I have tried placing the initialization of AvroParquetWriter at the open() method, but result still the same.

调试代码时，我确认执行了writer.write(element)并且element包含avro通用记录数据

When debugging the code, I confirm that writer.write(element) does executed and element contain the avro genericrecord data

流数据

BucketingSink<DataEventRecord> sink =
    new BucketingSink<DataEventRecord>("hdfs://localhost:9000/tmp/");

sink.setBucketer(new DateTimeBucketer<DataEventRecord>("yyyy-MM-dd--HHmm"));
sink.setWriter(new ParquetSinkWriter<DataEventRecord>());

ParquetSinkWriter

ParquetSinkWriter

import java.io.File;
import java.io.IOException;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.flink.streaming.connectors.fs.StreamWriterBase;
import org.apache.flink.streaming.connectors.fs.Writer;
import org.apache.parquet.avro.AvroParquetWriter;
import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
import com.any.DataEventRecord;

public class ParquetSinkWriter<T> extends StreamWriterBase<T> {

  private transient ParquetWriter<GenericRecord> writer;

  private Path path;
  private FileSystem fs;
  private final CompressionCodecName compressionCodecName = CompressionCodecName.SNAPPY;
  private final int blockSize = 256 * 1024 * 1024;
  private final int pageSize = 64 * 1024;


  @Override
  // workaround
  public void open(FileSystem fs, Path path) throws IOException {
    super.open(fs, path);
    this.path = path;
    this.fs = fs;
  }

  @Override
  public void write(T event) throws IOException {
    DataEventRecord element = (DataEventRecord) event;

    if (writer == null) {
      writer = new AvroParquetWriter<GenericRecord>(this.path, element.getSchema(), compressionCodecName, blockSize, pageSize);
    }

    if (writer != null) {
      GenericRecord datum = element.getRecord();
      writer.write(datum);
    }
  }

  @Override
  public void close() throws IOException {
    if (writer != null) {
      writer.close();
    }
    super.close();
  }

  @Override
  public Writer<T> duplicate() {
    return new ParquetSinkWriter<T>();
  }

}

推荐答案

直接实现Writer应该看起来像

import org.apache.flink.util.Preconditions;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.parquet.avro.AvroParquetWriter;
import org.apache.parquet.hadoop.ParquetWriter;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;

import java.io.IOException;

/**
 * Parquet writer.
 *
 * @param <T>
 */
public class ParquetSinkWriter<T extends GenericRecord> implements Writer<T> {

    private static final long serialVersionUID = -975302556515811398L;

    private final CompressionCodecName compressionCodecName = CompressionCodecName.SNAPPY;
    private final int pageSize = 64 * 1024;

    private final String schemaRepresentation;

    private transient Schema schema;
    private transient ParquetWriter<GenericRecord> writer;
    private transient Path path;

    private int position;

    public ParquetSinkWriter(String schemaRepresentation) {
        this.schemaRepresentation = Preconditions.checkNotNull(schemaRepresentation);
    }

    @Override
    public void open(FileSystem fs, Path path) throws IOException {
        this.position = 0;
        this.path = path;

        if (writer != null) {
            writer.close();
        }

        writer = createWriter();
    }

    @Override
    public long flush() throws IOException {
        Preconditions.checkNotNull(writer);
        position += writer.getDataSize();
        writer.close();
        writer = createWriter();

        return position;
    }

    @Override
    public long getPos() throws IOException {
        Preconditions.checkNotNull(writer);
        return position + writer.getDataSize();
    }

    @Override
    public void close() throws IOException {
        if (writer != null) {
            writer.close();
            writer = null;
        }
    }

    @Override
    public void write(T element) throws IOException {
        Preconditions.checkNotNull(writer);
        writer.write(element);
    }

    @Override
    public Writer<T> duplicate() {
        return new ParquetSinkWriter<>(schemaRepresentation);
    }

    private ParquetWriter<GenericRecord> createWriter() throws IOException {
        if (schema == null) {
            schema = new Schema.Parser().parse(schemaRepresentation);
        }

        return AvroParquetWriter.<GenericRecord>builder(path)
            .withSchema(schema)
            .withDataModel(new GenericData())
            .withCompressionCodec(compressionCodecName)
            .withPageSize(pageSize)
            .build();
    }
}

这篇关于使用自定义AvroParquetWriter的Flink BucketingSink创建空文件的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！