2

编辑：已解决！

我有来自多种语言的多个文本文件。我想使用 Apache Beam 为每一行添加一个语言标签。

例子：

文件text_en： This is a sentence.

文件text_de：Dies ist ein Satz.

我想要的是这样的：

zh: 这是一个句子。
de: Dies ist ein Satz。

我试过的：

我最初尝试只使用一个TextIO.Read.From(dataSetDirectory+"/*")并寻找一个看起来像.getSource(). 然而，这似乎并不存在。

接下来，我尝试像这样一个一个地读取每个文件：

File[] files = new File(datasetDirectory).listFiles();
PCollectionList<String> dataSet=null;
for (File f: files) {
   String language = f.getName();
   logger.debug(language);
   PCollection<String> newPCollection = p.apply(
            TextIO.Read.from(f.getAbsolutePath()))
               .apply(ParDo.of(new LanguageTagAdder(language)));

   if (dataSet==null) {
       dataSet=PCollectionList.of(newPCollection);
   } else {
       dataSet.and(newPCollection);
   }
}
PCollection<String> completeDataset= dataSet.apply(Flatten.<String>pCollections())

以这种方式读取文件效果很好，但是我的 DoFnLanguageTagAdder仅使用第一种语言进行初始化，因此所有文件都具有相同的添加语言。

LanguageTagAdder看起来像这样：

public class LanguageTagAdder
            extends DoFn<String,String> {

        private String language;
        public LanguageTagAdder(String language) {
            this.language=language;
        }
        @ProcessElement
        public void processElement(ProcessContext c) {
            c.output(language+c.element());
        }
    }

我意识到这种行为是有意和需要的，以便数据可以并行化，但我将如何解决我的问题？有没有Beam方法来解决它？

new LanguageTagAdderPS：我在第二次（使用第二种语言）创建时收到以下警告：

DEBUG   2016-12-05 17:09:55,070 [main] de.kdld16.hpi.FusionDataset  - en
DEBUG   2016-12-05 17:09:56,216 [main] de.kdld16.hpi.FusionDataset  - de
WARN    2016-12-05 17:09:56,219 [main] org.apache.beam.sdk.Pipeline  - Transform TextIO.Read2 does not have a stable unique name. This will prevent updating of pipelines.

编辑： 问题是线

dataSet.and(newPCollection);

它需要重写为：

dataSet=dataSet.and(newPCollection);

原来，dataSet 只包含第一个 File.... 难怪它们都有相同的语言标签！

4

1 回答 1

2

问题是线路

dataSet.and(newPCollection);

它需要重写为：

dataSet=dataSet.and(newPCollection);

原来，dataSet 只包含第一个文件。

于 2017-01-05T19:24:52.163 回答