<jupyter_start><jupyter_text>Enriching Data with moving statistics The bigquery cell magic allows for a rapid explorative development, but clutters the notebook with large result sets.<jupyter_code>%load_ext google.cloud.bigquery
%%bigquery
SELECT 
    DEP_T, AIRLINE, ARR, DEP_DELAY, ARR_DELAY
from `going-tfx.examples.ATL_JUNE_SIGNATURE` 
where date='2006-06-12'
order by dep_t limit 2<jupyter_output><empty_output><jupyter_text>---Doing it with ```datalab.bigquery``` doesn't display the large result set<jupyter_code>import google.datalab.bigquery as bq
samples=bq.Query("""
SELECT 
    DEP_T, AIRLINE, ARR, DEP_DELAY, ARR_DELAY
FROM 
    `going-tfx.examples.ATL_JUNE_SIGNATURE` 
WHERE 
    date='2006-06-12'
ORDER BY 
    dep_t
""").execute().result().to_dataframe()
two_records = samples[:2].to_dict(orient='records')
print("%s samples, showing first 2:" % len(samples))
two_records<jupyter_output>1075 samples, showing first 2:<jupyter_text>--- Beam Transform ```DoFn```s<jupyter_code>import apache_beam as beam
from apache_beam import window
from apache_beam.options.pipeline_options import PipelineOptions<jupyter_output><empty_output><jupyter_text>Add Timestamp<jupyter_code>class AddTimeStampDoFn(beam.DoFn):

    def __init__(self, offset, *args, **kwargs):
        self.offset = offset
        super(beam.DoFn, self).__init__(*args, **kwargs)
    
    def process(self, element):
        timestamp = (self.offset + 
                      (element['DEP_T'] // 100) * 3600 +
                      (element['DEP_T'] % 100) * 60)
        time_stamped = beam.window.TimestampedValue(element, timestamp)
        yield time_stamped<jupyter_output><empty_output><jupyter_text>Add hour of day<jupyter_code>class Add_HOD(beam.DoFn):
    def process(self, element):
        element=element.copy()
        dep_t = element['DEP_T']
        element['HOD'] = dep_t // 100
        yield element<jupyter_output><empty_output><jupyter_text>Key out DEP_DELAY for averaging<jupyter_code>class DEP_DELAY_by_HOD(beam.DoFn):
    def process(self, element):
        element=element.copy()
        yield element['HOD'], element['DEP_DELAY']<jupyter_output><empty_output><jupyter_text>Key the whole recordsKeying the records allows as to CoGroupByKey after the windowed statistics are available<jupyter_code>class Record_by_HOD(beam.DoFn):
    def process(self, element):
        element=element.copy()
        yield element['HOD'], element<jupyter_output><empty_output><jupyter_text>Unnest Unnest the resulting structure coming from ```CoGroupByKey``` to simple records<jupyter_code>class Flatten_EnrichedFN(beam.DoFn):
    def process(self, element):
        hod=element[0]
        avg=element[1]['avg'][0]
        cnt=element[1]['cnt'][0]
        records=element[1]['rec'][0]
        for record in records:
            record['CNT_BTH']=cnt
            record['AVG_DEP_DELAY_BTH']=avg
            yield record<jupyter_output><empty_output><jupyter_text>We'll add a timestamp as if the records were all today's records  Creating the pipeline<jupyter_code>import time
import datetime
OFFSET = int(time.time() // 86400 * 86400)
OFFSET
data = samples.to_dict(orient='records')
windowed = (
    data
    | "Add_timestamp" >> beam.ParDo(AddTimeStampDoFn(OFFSET))
    | "Add_HOD" >> beam.ParDo(Add_HOD())
    | "Window_1h" >> beam.WindowInto(window.FixedWindows(3600))
)
len(windowed)<jupyter_output><empty_output><jupyter_text>Counts by the hour<jupyter_code>records_by_hod = (
    windowed
    | "Record_by_HOD" >> beam.ParDo(Record_by_HOD())
    | "Group_by_HOD" >> beam.GroupByKey()
)
counts_by_hod = (
    records_by_hod
    | "Count" >> beam.CombineValues(beam.combiners.CountCombineFn())
)
counts_by_hod[:5]
#records_by_hod[2]<jupyter_output><empty_output><jupyter_text>Averages by the hour<jupyter_code>avgs_by_hod = (
    windowed    
    | "Make_HOD" >> beam.ParDo(DEP_DELAY_by_HOD())
    | "Avg_by_HOD" >> beam.CombinePerKey(beam.combiners.MeanCombineFn())
)
avgs_by_hod[:3]<jupyter_output><empty_output><jupyter_text>Co-Group and Flatten<jupyter_code>combined = ( {'cnt': counts_by_hod, 'avg': avgs_by_hod, 'rec': records_by_hod }
 | "Co_Group_HOD" >> beam.CoGroupByKey()
 | "Flatten" >> beam.ParDo(Flatten_EnrichedFN())
)
combined[:3]<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12535211267605634, 'fraction_numerical': 0.015258215962441314, 'mean_word_length': 4.877241379310345, 'pattern_counts': {'":': 0, '<': 48, '<?xml version=': 0, '>': 68, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '507401', 'license': 'Apache-2.0', 'n_tokens_mistral': 1606, 'n_tokens_neox': 1520, 'n_words': 267, 'path': 'kovalevskyi/Windowing_Tutorial.ipynb', 'repo_name': 'Project-Ellie/tutorials', 'size': '4260'}