title: Can I join two tables in Dataflow (Apache Beam) on multiple keys (join condition)?
body: <p>I have two tables like:</p>
  
  <pre><code>ID Name  Age
  1  Alex  20
  2  Sarah 21 and so on
  .....................
  ID Name  Marks
  1  Alex  80
  2  Sarah 78 and so on
  .....................
  </code></pre>
  
  <p>I want to join these two tables using Cloud Dataflow (Apache Beam) on more than one key (joining condition) i. e. ID and Name both are the common columns. How can I do so?</p>
  
  <p>I have tried joining it using one key (one common column) but I don't know how to use more than one key</p>
  
  <p>I have used this code as a reference:</p>
  
  <p><a href="https://github.com/GoogleCloudPlatform/professional-services/blob/master/examples/dataflow-python-examples/dataflow_python_examples/data_lake_to_mart.py" rel="nofollow noreferrer">https://github.com/GoogleCloudPlatform/professional-services/blob/master/examples/dataflow-python-examples/dataflow_python_examples/data_lake_to_mart.py</a></p>
  
  <pre class="lang-py prettyprint-override"><code>class JoinTables:
      def add_key_details(self, row, key_details):
          result = row.copy()
          try:
              result.update(key_details[row['name']])
          except KeyError as err:
              traceback.print_exc()
              logging.error("Name Not Found error: %s", err)
          return result
  
  def run(argv=None):
      jointables = JoinTables()
  
      table1= (p 
          | 'Read table1 details from BigQuery ' &gt;&gt; beam.io.Read(
               beam.io.BigQuerySource(
                    query='SELECT * FROM `dataset.table1`',
                    use_standard_sql=True
               )
          )
          | 'Key Details 1' &gt;&gt; beam.Map(lambda row: (row['name'], row))
      )
  
      table2 = (p 
          | 'Read table2 details from BigQuery ' &gt;&gt; beam.io.Read(
              beam.io.BigQuerySource(
                  query='SELECT * FROM `dataset.table2`',
                  use_standard_sql=True
              )
          )
          | 'Join data with side input 1' &gt;&gt; beam.Map(jointables.add_key_details, AsDict(table1))
      )
  </code></pre>