Как объединить данные в PCollection - Apache beam - PullRequest
Новая
       37

Как объединить данные в PCollection - Apache beam

1 голос
/ 14 июня 2019

Я ищу объединение данных в PCollection

вход представляет собой файл CSV 

customer id,customer name,transction amount,transaction type  
cust123,ravi,100,D  
cust123,ravi,200,D  
cust234,Srini,200,C  
cust444,shaker,500,D  
cust123,ravi,100,C  
cust123,ravi,300,C

O / p должно быть

После прочтения текстового файла в коллекцию объектов я хочу объединить, как показано на рисунке. 

cust123,ravi,300,D  
cust123,ravi,400,C  
cust234,Srini,200,C  
cust444,shaker,500,D

Pipeline pipeline = Pipeline.create(
   PipelineOptionsFactory.fromArgs(args).withValidation().create());

PCollection< Customer> pCollection =
   pipeline.apply("Read", TextIO.read().from("MyFile.csv"))
           .apply("splitData and store",
               ParDo.of(new TextTransform.SplitValues()))

1 Ответ

1 голос
/ 14 июня 2019

Если я правильно понимаю, вам нужно сложить суммы транзакций по группам по Customerid + тип транзакции. В этом случае вам нужно, с точки зрения высокого уровня:

  • назначить ключи для записей:
    • для этого можно использовать WithKeys PTransform, см. Документ ;
    • ключ зависит от вас, например, вы можете объединить идентификатор клиента с типом транзакции, например: csvField[0] + "," + csvField[3]
  • сгруппируйте записи по новому ключу, используя GroupByKey PTransform, см. Этот документ ;
  • выводом GBK будут коллекции записей с одним и тем же ключом, поэтому вам нужно будет применить ParDo, который будет принимать такую ​​коллекцию (все записи, принадлежащие одному клиенту и типу транзакции), суммируя сумма, выведите запись с суммой;

Последние два шага (GBK + ParDo), вероятно, можно заменить с помощью Combine.perKey() PTransform, который делает то же самое, но может быть оптимизирован во время выполнения. См. это и это для получения дополнительной информации.

Вы также можете посмотреть на Beam SQL, который позволил бы вам выразить ту же логику в SQL. См. этот документ для обзора Beam SQL. В этом случае вам нужно будет добавить ParDo, который преобразует записи CSV в ряд лучей перед применением SqlTransform.

...