Python Apache Beam Несколько выходов и обработка - PullRequest
Новая
       25

Python Apache Beam Несколько выходов и обработка

0 голосов
/ 29 августа 2018

Я все еще очень плохо знаком с концепциями Apache Beam и пытаюсь запустить задание в потоке данных Google со следующим потоком процессов: process_flow

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

Я написал следующий код: 

# List of values to filter by
x_list = [1, 2, 3]

with beam.Pipeline(options=PipelineOptions().from_dictionary(pipeline_params)) as p:
    # Read in newline JSON data - each line is a dictionary
    log_data = (
        p 
        | "Create " + input_file >> beam.io.textio.ReadFromText(input_file)
        | "Load " + input_file >> beam.FlatMap(lambda x: json.loads(x))
    )

    # For each value in x_list, filter log_data for dictionaries containing the value & write out to separate file
    for i in x_list:
        # Return dictionary if given key = value in filter
        filtered_log = log_data | "Filter_"+i >> beam.Filter(lambda x: x['key'] == i)
        # Do additional processing
        processed_log = process_pcoll(filtered_log, event)
        # Write final file
        output = (
            processed_log
            | 'Dump_json_'+filename >> beam.Map(json.dumps)
            | "Save_"+filename >> beam.io.WriteToText(output_fp+filename,num_shards=0,shard_name_template="")
        )

В настоящее время он обрабатывает только первое значение в списке. Я знаю, что мне, вероятно, придется использовать ParDo, но я не очень уверен, как учесть это в моем процессе.

Ценю любую помощь!

1 Ответ

0 голосов
/ 07 сентября 2018

Вы можете использовать TaggedOutput в Beam. Напишите функцию луча, которая будет помечать каждый элемент в pcollection. 

import uuid
import apache_beam as beam
import dateutil.parser
from apache_beam.pvalue import TaggedOutput

class TagData(beam.DoFn):
    def process(self, element):
        key = element.get('key')   
        yield TaggedOutput(key, element)



processed_tagged_log = processed_log | "tagged-data-by-key " >> beam.ParDo(TagData()).with_outputs(*x_list)

Теперь вы можете записать этот вывод в отдельный файл / таблицу 

# Write files to separate tables/files
    for key in x_list:
        processed_tagged_log[key] | "save file %s" % uuid.uuid4()>> beam.io.WriteToText(output_fp+key+filename,num_shards=0,shard_name_template="")

Надеюсь, это поможет!

источник: [https://beam.apache.org/documentation/sdks/pydoc/2.0.0/_modules/apache_beam/pvalue.html]

...