Google Dataflow: running dynamic query with BigQuery+Pub/Sub in Python












0















What I would like to do in the pipeline:




  1. Read from pub/sub (done)

  2. Transform this data to dictionary (done)

  3. Take the value of a specified key from the dict (done)


  4. Run a parametrized/dynamic query from BigQuery in which the where part should be like this:



    SELECT field1 FROM Table where field2 = @valueFromP/S





The pipeline



| 'Read from PubSub' >> beam.io.ReadFromPubSub(subscription='')
| 'String to dictionary' >> beam.Map(lambda s:data_ingestion.parse_method(s))
| 'BigQuery' >> <Here is where I'm not sure how to do it>


The normal way to read from BQ it would be like:



| 'Read' >> beam.io.Read(beam.io.BigQuerySource(
query="SELECT field1 FROM table where field2='string'", use_standard_sql=True))




I have read about parameterized queries but i'm not sure if this would work with apache beam.



It could be done using side inputs?



Which would be the best way to do this?





What I've tried:



def parse_methodBQ(input):
query=''SELECT field1 FROM table WHERE field1='%s' AND field2=True' % (input['field1'])'
return query


class ReadFromBigQuery(beam.PTransform):
def expand(self, pcoll):
return (
pcoll
| 'FormatQuery' >> beam.Map(parse_methodBQ)
| 'Read' >> beam.Map(lambda s: beam.io.Read(beam.io.BigQuerySource(query=s)))
)

with beam.Pipeline(options=pipeline_options) as p:
transform = (p | 'BQ' >> ReadFromBigQuery()


The result (why this?):



<Read(PTransform) label=[Read]>


The correct result should be like:



{u'Field1': u'string', u'Field2': Bool}




THE SOLUTION



In the pipeline:



| 'BQ' >> beam.Map(parse_method_BQ))


The function (using the BigQuery 0.25 API for dataflow)



def parse_method_BQ(input):
client = bigquery.Client()
QUERY = 'SELECT field1 FROM table WHERE field1='%s' AND field2=True' % (input['field1'])
client.use_legacy_sql = False
query_job = client.run_async_query(query=QUERY ,job_name='temp-query-job_{}'.format(uuid.uuid4())) # API request
query_job.begin()
while True:
query_job.reload() # Refreshes the state via a GET request.
if query_job.state == 'DONE':
if query_job.error_result:
raise RuntimeError(query_job.errors)
rows = query_job.results().fetch_data()
for row in rows:
if not (row[0] is None):
return input
time.sleep(1)









share|improve this question





























    0















    What I would like to do in the pipeline:




    1. Read from pub/sub (done)

    2. Transform this data to dictionary (done)

    3. Take the value of a specified key from the dict (done)


    4. Run a parametrized/dynamic query from BigQuery in which the where part should be like this:



      SELECT field1 FROM Table where field2 = @valueFromP/S





    The pipeline



    | 'Read from PubSub' >> beam.io.ReadFromPubSub(subscription='')
    | 'String to dictionary' >> beam.Map(lambda s:data_ingestion.parse_method(s))
    | 'BigQuery' >> <Here is where I'm not sure how to do it>


    The normal way to read from BQ it would be like:



    | 'Read' >> beam.io.Read(beam.io.BigQuerySource(
    query="SELECT field1 FROM table where field2='string'", use_standard_sql=True))




    I have read about parameterized queries but i'm not sure if this would work with apache beam.



    It could be done using side inputs?



    Which would be the best way to do this?





    What I've tried:



    def parse_methodBQ(input):
    query=''SELECT field1 FROM table WHERE field1='%s' AND field2=True' % (input['field1'])'
    return query


    class ReadFromBigQuery(beam.PTransform):
    def expand(self, pcoll):
    return (
    pcoll
    | 'FormatQuery' >> beam.Map(parse_methodBQ)
    | 'Read' >> beam.Map(lambda s: beam.io.Read(beam.io.BigQuerySource(query=s)))
    )

    with beam.Pipeline(options=pipeline_options) as p:
    transform = (p | 'BQ' >> ReadFromBigQuery()


    The result (why this?):



    <Read(PTransform) label=[Read]>


    The correct result should be like:



    {u'Field1': u'string', u'Field2': Bool}




    THE SOLUTION



    In the pipeline:



    | 'BQ' >> beam.Map(parse_method_BQ))


    The function (using the BigQuery 0.25 API for dataflow)



    def parse_method_BQ(input):
    client = bigquery.Client()
    QUERY = 'SELECT field1 FROM table WHERE field1='%s' AND field2=True' % (input['field1'])
    client.use_legacy_sql = False
    query_job = client.run_async_query(query=QUERY ,job_name='temp-query-job_{}'.format(uuid.uuid4())) # API request
    query_job.begin()
    while True:
    query_job.reload() # Refreshes the state via a GET request.
    if query_job.state == 'DONE':
    if query_job.error_result:
    raise RuntimeError(query_job.errors)
    rows = query_job.results().fetch_data()
    for row in rows:
    if not (row[0] is None):
    return input
    time.sleep(1)









    share|improve this question



























      0












      0








      0


      1






      What I would like to do in the pipeline:




      1. Read from pub/sub (done)

      2. Transform this data to dictionary (done)

      3. Take the value of a specified key from the dict (done)


      4. Run a parametrized/dynamic query from BigQuery in which the where part should be like this:



        SELECT field1 FROM Table where field2 = @valueFromP/S





      The pipeline



      | 'Read from PubSub' >> beam.io.ReadFromPubSub(subscription='')
      | 'String to dictionary' >> beam.Map(lambda s:data_ingestion.parse_method(s))
      | 'BigQuery' >> <Here is where I'm not sure how to do it>


      The normal way to read from BQ it would be like:



      | 'Read' >> beam.io.Read(beam.io.BigQuerySource(
      query="SELECT field1 FROM table where field2='string'", use_standard_sql=True))




      I have read about parameterized queries but i'm not sure if this would work with apache beam.



      It could be done using side inputs?



      Which would be the best way to do this?





      What I've tried:



      def parse_methodBQ(input):
      query=''SELECT field1 FROM table WHERE field1='%s' AND field2=True' % (input['field1'])'
      return query


      class ReadFromBigQuery(beam.PTransform):
      def expand(self, pcoll):
      return (
      pcoll
      | 'FormatQuery' >> beam.Map(parse_methodBQ)
      | 'Read' >> beam.Map(lambda s: beam.io.Read(beam.io.BigQuerySource(query=s)))
      )

      with beam.Pipeline(options=pipeline_options) as p:
      transform = (p | 'BQ' >> ReadFromBigQuery()


      The result (why this?):



      <Read(PTransform) label=[Read]>


      The correct result should be like:



      {u'Field1': u'string', u'Field2': Bool}




      THE SOLUTION



      In the pipeline:



      | 'BQ' >> beam.Map(parse_method_BQ))


      The function (using the BigQuery 0.25 API for dataflow)



      def parse_method_BQ(input):
      client = bigquery.Client()
      QUERY = 'SELECT field1 FROM table WHERE field1='%s' AND field2=True' % (input['field1'])
      client.use_legacy_sql = False
      query_job = client.run_async_query(query=QUERY ,job_name='temp-query-job_{}'.format(uuid.uuid4())) # API request
      query_job.begin()
      while True:
      query_job.reload() # Refreshes the state via a GET request.
      if query_job.state == 'DONE':
      if query_job.error_result:
      raise RuntimeError(query_job.errors)
      rows = query_job.results().fetch_data()
      for row in rows:
      if not (row[0] is None):
      return input
      time.sleep(1)









      share|improve this question
















      What I would like to do in the pipeline:




      1. Read from pub/sub (done)

      2. Transform this data to dictionary (done)

      3. Take the value of a specified key from the dict (done)


      4. Run a parametrized/dynamic query from BigQuery in which the where part should be like this:



        SELECT field1 FROM Table where field2 = @valueFromP/S





      The pipeline



      | 'Read from PubSub' >> beam.io.ReadFromPubSub(subscription='')
      | 'String to dictionary' >> beam.Map(lambda s:data_ingestion.parse_method(s))
      | 'BigQuery' >> <Here is where I'm not sure how to do it>


      The normal way to read from BQ it would be like:



      | 'Read' >> beam.io.Read(beam.io.BigQuerySource(
      query="SELECT field1 FROM table where field2='string'", use_standard_sql=True))




      I have read about parameterized queries but i'm not sure if this would work with apache beam.



      It could be done using side inputs?



      Which would be the best way to do this?





      What I've tried:



      def parse_methodBQ(input):
      query=''SELECT field1 FROM table WHERE field1='%s' AND field2=True' % (input['field1'])'
      return query


      class ReadFromBigQuery(beam.PTransform):
      def expand(self, pcoll):
      return (
      pcoll
      | 'FormatQuery' >> beam.Map(parse_methodBQ)
      | 'Read' >> beam.Map(lambda s: beam.io.Read(beam.io.BigQuerySource(query=s)))
      )

      with beam.Pipeline(options=pipeline_options) as p:
      transform = (p | 'BQ' >> ReadFromBigQuery()


      The result (why this?):



      <Read(PTransform) label=[Read]>


      The correct result should be like:



      {u'Field1': u'string', u'Field2': Bool}




      THE SOLUTION



      In the pipeline:



      | 'BQ' >> beam.Map(parse_method_BQ))


      The function (using the BigQuery 0.25 API for dataflow)



      def parse_method_BQ(input):
      client = bigquery.Client()
      QUERY = 'SELECT field1 FROM table WHERE field1='%s' AND field2=True' % (input['field1'])
      client.use_legacy_sql = False
      query_job = client.run_async_query(query=QUERY ,job_name='temp-query-job_{}'.format(uuid.uuid4())) # API request
      query_job.begin()
      while True:
      query_job.reload() # Refreshes the state via a GET request.
      if query_job.state == 'DONE':
      if query_job.error_result:
      raise RuntimeError(query_job.errors)
      rows = query_job.results().fetch_data()
      for row in rows:
      if not (row[0] is None):
      return input
      time.sleep(1)






      python google-bigquery google-cloud-dataflow apache-beam google-cloud-pubsub






      share|improve this question















      share|improve this question













      share|improve this question




      share|improve this question








      edited Nov 22 '18 at 7:19







      IoT user

















      asked Nov 20 '18 at 11:00









      IoT userIoT user

      13512




      13512
























          1 Answer
          1






          active

          oldest

          votes


















          1














          You can read the whole table or use a string query.



          I understand that you will use the parse_methodBQ method to customize the query as needed. As this method returns a query, you can call it with BigQuerySource. The rows are in dictionary.



          | 'QueryTable' >> beam.Map(beam.io.BigQuerySource(parse_methodBQ))
          # Each row is a dictionary where the keys are the BigQuery columns
          | 'Read' >> beam.Map(lambda s: s['data'])


          Further more, you can avoid having to customize the query and use a filter method



          Regarding the side inputs, review this example from the cookbook to have a better view on how to use them.






          share|improve this answer
























          • Thank for your answer. I did it using beam.Map so it should be like the solution you propose. I will add the solution to the main post.

            – IoT user
            Nov 22 '18 at 7:15













          • Which would be better for max performance: using filter data or using a parse_method like I did?

            – IoT user
            Nov 22 '18 at 9:41











          • Depending on your use case, if you will you are using multiple custom queries it may be better to have all the table an use a filter but if you are only doing a few custom queries having all the table may be useless.

            – Nathan Nasser
            Nov 22 '18 at 19:43











          Your Answer






          StackExchange.ifUsing("editor", function () {
          StackExchange.using("externalEditor", function () {
          StackExchange.using("snippets", function () {
          StackExchange.snippets.init();
          });
          });
          }, "code-snippets");

          StackExchange.ready(function() {
          var channelOptions = {
          tags: "".split(" "),
          id: "1"
          };
          initTagRenderer("".split(" "), "".split(" "), channelOptions);

          StackExchange.using("externalEditor", function() {
          // Have to fire editor after snippets, if snippets enabled
          if (StackExchange.settings.snippets.snippetsEnabled) {
          StackExchange.using("snippets", function() {
          createEditor();
          });
          }
          else {
          createEditor();
          }
          });

          function createEditor() {
          StackExchange.prepareEditor({
          heartbeatType: 'answer',
          autoActivateHeartbeat: false,
          convertImagesToLinks: true,
          noModals: true,
          showLowRepImageUploadWarning: true,
          reputationToPostImages: 10,
          bindNavPrevention: true,
          postfix: "",
          imageUploader: {
          brandingHtml: "Powered by u003ca class="icon-imgur-white" href="https://imgur.com/"u003eu003c/au003e",
          contentPolicyHtml: "User contributions licensed under u003ca href="https://creativecommons.org/licenses/by-sa/3.0/"u003ecc by-sa 3.0 with attribution requiredu003c/au003e u003ca href="https://stackoverflow.com/legal/content-policy"u003e(content policy)u003c/au003e",
          allowUrls: true
          },
          onDemand: true,
          discardSelector: ".discard-answer"
          ,immediatelyShowMarkdownHelp:true
          });


          }
          });














          draft saved

          draft discarded


















          StackExchange.ready(
          function () {
          StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53391541%2fgoogle-dataflow-running-dynamic-query-with-bigquerypub-sub-in-python%23new-answer', 'question_page');
          }
          );

          Post as a guest















          Required, but never shown

























          1 Answer
          1






          active

          oldest

          votes








          1 Answer
          1






          active

          oldest

          votes









          active

          oldest

          votes






          active

          oldest

          votes









          1














          You can read the whole table or use a string query.



          I understand that you will use the parse_methodBQ method to customize the query as needed. As this method returns a query, you can call it with BigQuerySource. The rows are in dictionary.



          | 'QueryTable' >> beam.Map(beam.io.BigQuerySource(parse_methodBQ))
          # Each row is a dictionary where the keys are the BigQuery columns
          | 'Read' >> beam.Map(lambda s: s['data'])


          Further more, you can avoid having to customize the query and use a filter method



          Regarding the side inputs, review this example from the cookbook to have a better view on how to use them.






          share|improve this answer
























          • Thank for your answer. I did it using beam.Map so it should be like the solution you propose. I will add the solution to the main post.

            – IoT user
            Nov 22 '18 at 7:15













          • Which would be better for max performance: using filter data or using a parse_method like I did?

            – IoT user
            Nov 22 '18 at 9:41











          • Depending on your use case, if you will you are using multiple custom queries it may be better to have all the table an use a filter but if you are only doing a few custom queries having all the table may be useless.

            – Nathan Nasser
            Nov 22 '18 at 19:43
















          1














          You can read the whole table or use a string query.



          I understand that you will use the parse_methodBQ method to customize the query as needed. As this method returns a query, you can call it with BigQuerySource. The rows are in dictionary.



          | 'QueryTable' >> beam.Map(beam.io.BigQuerySource(parse_methodBQ))
          # Each row is a dictionary where the keys are the BigQuery columns
          | 'Read' >> beam.Map(lambda s: s['data'])


          Further more, you can avoid having to customize the query and use a filter method



          Regarding the side inputs, review this example from the cookbook to have a better view on how to use them.






          share|improve this answer
























          • Thank for your answer. I did it using beam.Map so it should be like the solution you propose. I will add the solution to the main post.

            – IoT user
            Nov 22 '18 at 7:15













          • Which would be better for max performance: using filter data or using a parse_method like I did?

            – IoT user
            Nov 22 '18 at 9:41











          • Depending on your use case, if you will you are using multiple custom queries it may be better to have all the table an use a filter but if you are only doing a few custom queries having all the table may be useless.

            – Nathan Nasser
            Nov 22 '18 at 19:43














          1












          1








          1







          You can read the whole table or use a string query.



          I understand that you will use the parse_methodBQ method to customize the query as needed. As this method returns a query, you can call it with BigQuerySource. The rows are in dictionary.



          | 'QueryTable' >> beam.Map(beam.io.BigQuerySource(parse_methodBQ))
          # Each row is a dictionary where the keys are the BigQuery columns
          | 'Read' >> beam.Map(lambda s: s['data'])


          Further more, you can avoid having to customize the query and use a filter method



          Regarding the side inputs, review this example from the cookbook to have a better view on how to use them.






          share|improve this answer













          You can read the whole table or use a string query.



          I understand that you will use the parse_methodBQ method to customize the query as needed. As this method returns a query, you can call it with BigQuerySource. The rows are in dictionary.



          | 'QueryTable' >> beam.Map(beam.io.BigQuerySource(parse_methodBQ))
          # Each row is a dictionary where the keys are the BigQuery columns
          | 'Read' >> beam.Map(lambda s: s['data'])


          Further more, you can avoid having to customize the query and use a filter method



          Regarding the side inputs, review this example from the cookbook to have a better view on how to use them.







          share|improve this answer












          share|improve this answer



          share|improve this answer










          answered Nov 21 '18 at 21:49









          Nathan NasserNathan Nasser

          4069




          4069













          • Thank for your answer. I did it using beam.Map so it should be like the solution you propose. I will add the solution to the main post.

            – IoT user
            Nov 22 '18 at 7:15













          • Which would be better for max performance: using filter data or using a parse_method like I did?

            – IoT user
            Nov 22 '18 at 9:41











          • Depending on your use case, if you will you are using multiple custom queries it may be better to have all the table an use a filter but if you are only doing a few custom queries having all the table may be useless.

            – Nathan Nasser
            Nov 22 '18 at 19:43



















          • Thank for your answer. I did it using beam.Map so it should be like the solution you propose. I will add the solution to the main post.

            – IoT user
            Nov 22 '18 at 7:15













          • Which would be better for max performance: using filter data or using a parse_method like I did?

            – IoT user
            Nov 22 '18 at 9:41











          • Depending on your use case, if you will you are using multiple custom queries it may be better to have all the table an use a filter but if you are only doing a few custom queries having all the table may be useless.

            – Nathan Nasser
            Nov 22 '18 at 19:43

















          Thank for your answer. I did it using beam.Map so it should be like the solution you propose. I will add the solution to the main post.

          – IoT user
          Nov 22 '18 at 7:15







          Thank for your answer. I did it using beam.Map so it should be like the solution you propose. I will add the solution to the main post.

          – IoT user
          Nov 22 '18 at 7:15















          Which would be better for max performance: using filter data or using a parse_method like I did?

          – IoT user
          Nov 22 '18 at 9:41





          Which would be better for max performance: using filter data or using a parse_method like I did?

          – IoT user
          Nov 22 '18 at 9:41













          Depending on your use case, if you will you are using multiple custom queries it may be better to have all the table an use a filter but if you are only doing a few custom queries having all the table may be useless.

          – Nathan Nasser
          Nov 22 '18 at 19:43





          Depending on your use case, if you will you are using multiple custom queries it may be better to have all the table an use a filter but if you are only doing a few custom queries having all the table may be useless.

          – Nathan Nasser
          Nov 22 '18 at 19:43




















          draft saved

          draft discarded




















































          Thanks for contributing an answer to Stack Overflow!


          • Please be sure to answer the question. Provide details and share your research!

          But avoid



          • Asking for help, clarification, or responding to other answers.

          • Making statements based on opinion; back them up with references or personal experience.


          To learn more, see our tips on writing great answers.




          draft saved


          draft discarded














          StackExchange.ready(
          function () {
          StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53391541%2fgoogle-dataflow-running-dynamic-query-with-bigquerypub-sub-in-python%23new-answer', 'question_page');
          }
          );

          Post as a guest















          Required, but never shown





















































          Required, but never shown














          Required, but never shown












          Required, but never shown







          Required, but never shown

































          Required, but never shown














          Required, but never shown












          Required, but never shown







          Required, but never shown







          Popular posts from this blog

          Guess what letter conforming each word

          Port of Spain

          Run scheduled task as local user group (not BUILTIN)