google-cloud-dataflow - 使用 Dataflow 读取 CSV header

Question

我有一个 CSV 文件，但我提前不知道列名。我需要在 Google Dataflow 中进行一些转换后以 JSON 格式输出数据。

获取标题行并将标签渗透到所有行的最佳方法是什么？

例如:

a,b,c
1,2,3
4,5,6

...变成(大约):

{a:1, b:2, c:3}
{a:4, b:5, c:6}

Answer 1

您应该实现自定义 FileBasedSource (类似于 TextIO.TextSource )，它将读取第一行并存储标题数据

    @Override
    protected void startReading(final ReadableByteChannel channel)
    throws IOException {
        lineReader = new LineReader(channel);

        if (lineReader.readNextLine()) {
            final String headerLine = lineReader.getCurrent().trim();
            header = headerLine.split(",");
            readingStarted = true;
        }
    }

后者，在读取其他行时将其添加到当前行数据中:

    @Override
    protected boolean readNextRecord() throws IOException {
        if (!lineReader.readNextLine()) {
            return false;
        }

        final String line = lineReader.getCurrent();
        final String[] data = line.split(",");

        // assumes all lines are valid
        final StringBuilder record = new StringBuilder();
        for (int i = 0; i < header.length; i++) {
            record.append(header[i]).append(":").append(data[i]).append(", ");
        }

        currentRecord = record.toString();
        return true;
    }

我已经实现了一个快速(完整)的解决方案，可在 github 上获得.我还添加了一个数据流单元测试来演示阅读:

@Test
public void test_reading() throws Exception {
    final File file =
            new File(getClass().getResource("/sample.csv").toURI());
    assertThat(file.exists()).isTrue();

    final Pipeline pipeline = TestPipeline.create();

    final PCollection<String> output =
            pipeline.apply(Read.from(CsvWithHeaderFileSource.from(file.getAbsolutePath())));

    DataflowAssert
            .that(output)
            .containsInAnyOrder("a:1, b:2, c:3, ", "a:4, b:5, c:6, ");

    pipeline.run();
}

哪里 sample.csv有以下内容:

a,b,c
1,2,3
4,5,6