Left join in apache beam

user_2717021 0
Viewed 3757

Which is the better way to left join following Pcollection in apache beam?

pcoll1 = [('key1', [[('a', 1)],[('b', 2)], [('c', 3)], [('d', 4)],[('e', 5)], [('f', 6)]]), ('key2',[[('a', 12)],[('b', 21)], [('c', 13)]]), ('key3',[[('a', 21)],[('b', 23)], [('c', 31)]])]
pcoll2 = [('key1', [[('x', 10)]]), ('key2', [[('x', 20)]])]

Expected outpus is 

[('a', 1), ('x', 10)]
[('b', 2), ('x', 10)] 
[('c', 3), ('x', 10)] 
[('d', 4), ('x', 10)]
[('e', 5), ('x', 10)] 
[('f', 6), ('x', 10)]
[('a', 12), ('x', 20)]
[('b', 21), ('x', 20)] 
[('c', 13), ('x', 20)]
[('a', 21)]
[('b', 23)]
[('c', 31)]

I have implemented a left joiner using CoGroupByKey() and Pardo(). Is there any other method to implement left joiner in beam Python SDK?

left_joined = (
    {'left': pcoll1, 'right': pcoll2}
    | 'LeftJoiner: Combine' >> beam.CoGroupByKey()
    | 'LeftJoiner: ExtractValues' >> beam.Values()
    | 'LeftJoiner: JoinValues' >> beam.ParDo(LeftJoinerFn())
)


class LeftJoinerFn(beam.DoFn):

    def __init__(self):
        super(LeftJoinerFn, self).__init__()

    def process(self, row, **kwargs):

        left = row['left']
        right = row['right']

        if left and right:
            for each in left:
                yield each + right[0]

        elif left:
            for each in left:
                yield each
apache-beam python
3 Answers

You can use the follwing code for using side inputs for the right side of the join, assuming the right side is always going to have one element mapped to each key which means that it is always much smaller in size than the left pcollection.

Also, if your pcollection is created by reading from an external source instead of an in-memory array, you will need to pass right_list=beam.pvalue.asList(pcoll2) instead of right_list=pcoll2 to the ParDo . Check here for more info

class LeftJoinerFn(beam.DoFn):

    def __init__(self):
        super(LeftJoinerFn, self).__init__()

    def process(self, row, **kwargs):

        right_dict = dict(kwargs['right_list'])
        left_key = row[0]

        if left_key in right_dict:
            for each in row[1]:
                yield each + right_dict[left_key]

        else:
            for each in row[1]:
                yield each

class Display(beam.DoFn):
    def process(self, element):
        LOG.info(str(element))
        yield element

p = beam.Pipeline(options=pipeline_options)

pcoll1 = [('key1', [[('a', 1)],[('b', 2)], [('c', 3)], [('d', 4)],[('e', 5)], [('f', 6)]]), \
        ('key2',[[('a', 12)],[('b', 21)], [('c', 13)]]), \
        ('key3',[[('a', 21)],[('b', 23)], [('c', 31)]])\
        ]
pcoll2 = [('key1', [[('x', 10)]]), ('key2', [[('x', 20)]])]


left_joined = (
    pcoll1 
    | 'LeftJoiner: JoinValues' >> beam.ParDo(LeftJoinerFn(), right_list=pcoll2)
    | 'Display' >> beam.ParDo(Display())
)
p.run()
user_36483180

left.txt

149633CM,Marco,10

212539MU,Rebekah,10

231555ZZ,Itoe,10

right.txt

149633CM,Australia

212539MU,India

Code for left Join:

from apache_beam.io.gcp.internal.clients import bigquery
import apache_beam as beam

def retTuple(element):
  
  thisTuple=element.split(',')
  return (thisTuple[0],thisTuple[1:])



def jstr(j):
    import datetime
    jlist=[]
    for k in ((j[1]['left_data'])):
        if len((j[1]['right_data']))==0:
            id,name,rank=([j[0]]+k)
            json_str={ "id":id,"name":name,"rank":rank}
            jlist.append(json_str)
        else:
            for l in ((j[1]['right_data'])):
                
        # print(([j[0]]+k+l))
                id,name,rank,country=([j[0]]+k+l)
            
                json_str={ "id":id,"name":name,"rank":rank,"country":country }
                jlist.append(json_str)
    return jlist
            
table_spec = 'project:dataset.table_name'
table_schema = 'id:STRING,name:STRING,rank:INTEGER,country:STRING'   
gcs='gs://dataflow4bigquery/temp'
    


p1 = beam.Pipeline()

# Apply a ParDo to the PCollection "words" to compute lengths for each word.
left_rows = ( 
                p1
                | "Reading File 1" >> beam.io.ReadFromText('left.txt')
                | 'Pair each employee with key' >> beam.Map(retTuple)          # {149633CM : [Marco,10]}
    
               )


right_rows = ( 
                p1
                | "Reading File 2" >> beam.io.ReadFromText('right.txt') 
                | 'Pair each country with key' >> beam.Map(retTuple)                # {149633CM : [9876843261,New York]}
               )


results = ({'left_data': left_rows, 'right_data': right_rows} 
           
           | beam.CoGroupByKey()
           | beam.FlatMap(jstr)
           | beam.io.WriteToBigQuery(
               custom_gcs_temp_location=gcs,
            table=table_spec,
            schema=table_schema,
            write_disposition=beam.io.BigQueryDisposition.WRITE_TRUNCATE,
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
       

)
)         
 
p1.run().wait_until_finish()
user_119275820
Related