我是云数据流和 Java 的新手,所以我希望这是正确的问题。
我有一个 csv 文件,其中有 n 个列和行,可以是字符串、整数或时间戳。我需要为每一列创建一个新的 PCollection 吗?
我在示例中找到的大多数文档都类似于:
PCollection<String> data = p.apply(TextIO.Read.from("gs://abc/def.csv"));
但对我来说,将整个 csv 文件作为字符串导入是没有意义的。我在这里缺少什么?我应该如何设置我的 PCollections?
line.split(",");
如果行数据如下所示,则 String.split 没有意义:
a,b,c,"我们有一个包含逗号的字符串",d,e
处理 csv 数据的一种属性方法是导入 csv 库:
<dependency>
<groupId>com.opencsv</groupId>
<artifactId>opencsv</artifactId>
<version>3.7</version>
</dependency>
并在 ParDo 中使用以下代码:
public void processElement(ProcessContext c) throws IOException {
String line = c.element();
CSVParser csvParser = new CSVParser();
String[] parts = csvParser.parseLine(line);
}
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)