我正在使用第二代云函数来触发数据流作业。数据流模板基本上是从云存储中读取拼花文件并将数据加载到bigTable上。以下是代码和包的详细信息

import os
import datetime
import logging
from configparser import ConfigParser
import apache_beam as beam
from google.cloud.bigtable import Client
from google.cloud.bigtable.row import DirectRow
from apache_beam.options.pipeline_options import PipelineOptions
from google.cloud import bigtable
from google.cloud.bigtable import column_family
from google.cloud.bigtable import row_filters

from apache_beam.io.gcp.bigtableio import WriteToBigTable

logger = logging.getLogger()
logger.setLevel(logging.INFO)

config_object = ConfigParser()
config_object.read("config.ini")

project_id = config_object["uprn"]["project_id"]
instance_id = config_object["uprn"]["instance_id"]
table_id = config_object["uprn"]["table_id"]
column_family_id = config_object["uprn"]["column_family_id"]
#input_columns = config_object["uprn"]["input_columns"]
timestamp = datetime.datetime(1970, 1, 1)
logging.info("--Starting..")

#client = bigtable.Client(project=project_id, admin=True)
#instance = client.instance(instance_id)
#table = instance.table(table_id)

def big_table_load(ele):
    try:
        rows = []
        column_names = list(ele.keys())
        row_key = str(str(ele['uprn'])).encode()
        logging.info("--row_key "+str(row_key))
        row = DirectRow(row_key)

        for key in column_names:
            row.set_cell(
                column_family_id, key, str(ele[key]).encode('utf-8'), timestamp=timestamp
            )
        rows.append(row)
        return rows
    except Exception as e:
        logging.info("Error encountered for row_key " + str(row_key) + " with error message "+ str(e))

def find_err_file():
    filename_err = user_options.efilename.get()
    return filename_err


class UserOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        parser.add_value_provider_argument('--input_location',
                                           default='gs://my-proj-dev-local-landing-zone/mock_data/*'
                                           )


pipeline_options = PipelineOptions()
user_options = pipeline_options.view_as(UserOptions)


def run():
    try:
        with beam.Pipeline(options=pipeline_options) as p:
            records = (p | 'Read' >> beam.io.ReadFromParquet(user_options.input_location)
                       | 'Format Rows' >> beam.ParDo(big_table_load)
                       | WriteToBigTable(
                        project_id=project_id,
                        instance_id=instance_id,
                        table_id=table_id
                    )
                       )
    except Exception as e:
        logging.info(e)
        raise e


if __name__ == '__main__':
    run()

要求. txt

google-cloud-bigtable==1.7.0
apache-beam[gcp]==2.39.0