markcutajar

Running an Apache Beam pipeline in a Windows Local Spark Cluster

This file will include details on attempting to run an apache beam pipeline on a windows spark setup. Unlike many other techies out there, i am a firm Microsoft fan. I do believe they build quality products and over the last few years since Satya Nadella was handed the position of CEO, we're seeing massive strides on Microsoft products.

To that point, all my machines at home are windows based including my processing machine which i mosty use to run models or occasionally play a few games. A couple of weeks ago i decided to build a simple Apache Beam pipeline to process forex tick data into windowed candle stick form. Now this can be easily done in pandas if the data was smaller, having a 7GB tick data file, in order to do it using pandas we'd have to split the file into day or maybe week segments and process them one by one. In order to obviously avoid the simple way out, i decided to attempt to build a beam pipeline as it is promised to pr

markcutajar

rows = self.pipeline | 'Read data file' >> ReadFromText(
    self.input_file_pattern,
    compression_type=_compression
)

markcutajar

if self.input_file_type == 'csv':
    rows = rows | 'Convert CSV to tick elements' >> beam.ParDo(
        ParseDataRows(),
        headers=self.input_columns
    )
else:
    rows = rows | 'Convert JSON to tick elements' >> beam.Map(json.loads)

markcutajar

rows = rows | 'Convert to timestamp field' >> beam.ParDo(
    ToTimestamp(),
    string_format=self.TICK_DATA_TIMESTAMP_FORMAT,
    datetime_key=self.input_columns[0],
    timestamp_key='timestamp'
)

rows = rows | 'Convert to datetime object' >> beam.ParDo(
    AddTimestamp(),
    timestamp_key='timestamp'

markcutajar

class AddTimestamp(beam.DoFn):
    """ParDo take a timestamp column and assign it 
		to row to produce a timestamped value
		"""
    def process(self, element, timestamp_key, delete_key=False, **kwargs):
        """
        :param element: Element being processed
        :param timestamp_key: Field containing the timestamp
        :param delete_key: Whether to delete the original timestamp or not
        :return: Elements with a datetime filed

markcutajar

...
yield {
    **element, 
    time_group_key: element['timestamp'] // window_size
}

markcutajar

def map_elements(self, data):
      """Map function to create key:value pairs to run CombinePerKey function.
      :param data: PCollection being processed with time_group_key column and
          instrument_column if set to use.
      :return: PCollection with mapped data
      """
      action_name = 'Resampler - Map data'
      if self.instrument_column:
          return data | action_name >> beam.Map(
              lambda x: ((x['time_group_key'], x[self.instrument_column]), x)

markcutajar

class CustomCombineFn(beam.CombineFn):
  def create_accumulator(self):
    return # create accumulator here

  def add_input(self, current_accumulation, input):
    return # perform accumulation here

  def merge_accumulators(self, accumulators):
    return # merge accumulators running un different nodes

markcutajar

@staticmethod
def get_value(item):
    """Function to calculate the value of the element.
    This is done so we can easily change how the value is calculated,
    by changing it in this one place.
    :param item: Tick item
    :return: Value of tick item
    """
    return item['ask']

markcutajar

class CustomTestCase(unittest.TestCase):
    def setUp(self):
        pipeline_options = PipelineOptions(['--runner=DirectRunner'])
        pipeline_options.view_as(SetupOptions).save_main_session = True
        self.pipeline = beam.Pipeline(options=pipeline_options)


class TestFunctionX(CustomTestCase):
    def test_run(self):
        with self.pipeline: