Google DataFlow, how to wait for external webhook

Question

I have a code that reads an Xlsx file, and for each line, do a process on a specific column.

The problem is related to the "Transform" part of the Dataflow. I implemented a specific method that get the value sent from the reader, and this data is sent to an outside server. This outside server process the data (could takes minutes), then do a POST request with the result. (the URL for the POST request is specified in the original request.

My questions is the following : how can I make my ParDo method be notified when the outside process is done (outside callback) ?

Here's my code so far :

import logging, argparse
import apache_beam as beam
from apache_beam.io import gcsio
from apache_beam.utils.options import PipelineOptions

from openpyxl import load_workbook


# @See https://cloud.google.com/dataflow/model/custom-io-python#ptransform-wrappers
class FileReader():
    """A file reader implementation"""

    def __init__(self, path, *args, **kwargs):
        self.path = path

    def reader(self):
        return XlsxFileReader(self.path)


class XlsxFileReader():
    """The Xlsx file reader"""
    def __init__(self, path):
        self.path = path

    def _clean_value(self, value):
        if value is None:
            return None

        value = unicode(value)

        try:
            value = value.encode('utf-8')
        except UnicodeEncodeError:
            pass

        return value

    def __iter__(self):
        wb = load_workbook(filename=self.file, read_only=True)
        sheet_name = wb.get_sheet_names()[0]
        ws = wb[sheet_name]
        for line, row in enumerate(ws.rows):
            cell_value = self._clean_value(row[0].value)
            if cell_value is not None and cell_value.find('@') > 0:
                yield cell_value, line
                break

    def __enter__(self):
        self.file = gcsio.GcsIO().open(self.path, 'r')
        return self

    def __exit__(self, *args, **kwargs):
        self.file.close()


class ComputeWordLengthFn(beam.DoFn):
    def process(self, context):
        # Here, what I would need is send a request to an external API, that returns the result to the `callback` parameter.
        # I know how to do that using requests
        #
        # ***********************************************************
        # ---> BUT HOW can I know when that external service has done with my data and called back my `callback` url?
        # ***********************************************************
        yield context.element[0] is done once external service has made a request to the `callback` url on my instance.


def run(argv=None):
    parser = argparse.ArgumentParser()
    parser.add_argument(
        '--input',
        dest='input',
        default='gs://norbert-verify-staging/growthlist.xlsx',
        help='Input file to process.'
    )
    parser.add_argument(
        '--output',
        dest='output',
        required=True,
        help='Output file to write results to.'
    )
    known_args, pipeline_args = parser.parse_known_args(argv)

    pipeline_options = PipelineOptions(pipeline_args)
    p = beam.Pipeline(options=pipeline_options)

    p | 'read' >> beam.io.Read(FileReader(known_args.input)) \
      | 'verify' >> beam.ParDo(ComputeWordLengthFn()) \
      | 'write' >> beam.io.Write(beam.io.TextFileSink(known_args.output))

    p.run()


if __name__ == '__main__':
    logging.getLogger().setLevel(logging.INFO)
    run()

I hope it's clear, let me know if you need more details.

Answer 1

Not sure if I fully understood your question but seems like you are asking if Beam offers a way for a DoFn.process() method to be called once a given callback has been invoked. Currently Beam does not offer such a feature.

What you can do here is to wait within the ComputeWordLengthFn.process() method till the request for a particular element has been completed (Exact way to perform that wait depends on the external API).

Please let me know if I misunderstood your question.