从Bigquery中滤出记录

Question

null

with beam.Pipeline(options=pipeline_args) as p:
        #read the dataset from bigquery
        query_top_30_items = (
            p 
            | 'GetTopItemNumbers' >> beam.io.ReadFromBigQuery(
                query="""SELECT item_number, COUNT(item_number) AS freq_count FROM 
                [bigquery-public-data.iowa_liquor_sales.sales] GROUP BY item_number 
                ORDER BY freq_count DESC
                LIMIT 30"""
            )
            | 'ReadItemNumbers' >> beam.Map(lambda elem: elem['item_number'])
            | 'ItemNumberAsList' >> beam.combiners.ToList()
        )


        query_top_30_stores = (
            p
            |
            'GetTopStores' >> beam.io.ReadFromBigQuery(
                query = """SELECT store_number, COUNT(store_number) AS store_count
                 FROM [bigquery-public-data.iowa_liquor_sales.sales] GROUP BY
                 store_number ORDER BY store_count DESC LIMIT 30"""
            )
            | 'ReadStoreValues' >> beam.Map(lambda elem:elem['store_number'])
            | 'StoreValuesAsList' >> beam.combiners.ToList()
        )

        query_whole_table = (
            (query_top_30_items, query_top_30_stores)
            |'ReadTable' >> beam.io.ReadFromBigQuery(
                query="""SELECT item_number, store_number, bottles_sold,
                    state_bottle_retail  FROM [bigquery-public-data.iowa_liquor_sales.sales]""")
            | 'FilterByItems' >> beam.Filter(lambda row:row['item_number'] in query_top_30_items)
            | 'FilterByStore' >> beam.Filter(lambda row:row['store_number'] in query_top_30_stores)
        )

temp_location=pcoll.pipeline.options.view_as（Traceback（最近的调用最后）：文件“run.py”,第113行，在run（）文件“run.py”中，第100行，在run“filterbystore”>>beam.filter（lambda row：row['store_number']，在query_top_30_stores中）文件“lib/python3.7/site-packages/apache_beam/transforms/ptransform.py”，第573行，在ror result=p.applice(self,pvalueish,label）文件“/library/frameworks/python.framework/versions/3.7/lib/python3.7/site-packages/apache_beam/pipeline.py”中，第646行，在apply return self.applice(transform，pvalueish,label）文件“ish,self._options)文件”/Library/frameworks/python.framework/versions/3.7/lib/python3.7/site-packages/apache_beam/runners/runner.py“，第188行，在apply return m（转换，输入，选项）文件”/library/frameworks/python.framework/versions/3.7/lib/python3.7/site-packon.framework/versions/3.7/lib/python3.7/site-packages/apache_beam/runners/runner.py“中，第218行，在=pcoll.pipeline.options.view_as(attributeError：”tuple“对象没有属性”pipeline“

因为我是Beam的新手，所以代码没有那么优化。请让我知道如果我可以进一步优化这个代码。

感谢您的时间和帮助！

Answer 1

对函数应用筛选条件在管道中不起作用。您有两个相同的选择：-

1. 在管道内应用筛选条件。
2. 在BQ-SQL上应用筛选条件。

函数上的筛选条件对于函数返回给调用函数的内容将是不明确的。因此，修改您的代码，将筛选条件应用到上面突出显示的两个位置中的任何一个。