使用Dataflow向Cloud BigTable写入数据时遇not_found错误求助

问题背景 #

我通过第二代Cloud Function触发Dataflow任务，该任务从Cloud Storage读取Parquet文件并将数据加载至Cloud BigTable，但写入时出现not_found错误，请求协助排查。

代码详情 #

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()

依赖包信息（requirements.txt） #

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

报错详情 #

Error processing instruction process_bundle-4225915941562411087-3. Original traceback is Traceback (most recent call last): File "apache_beam/runners/common.py", line 1232, in apache_beam.runners.common.DoFnRunner._invoke_bundle_method File "apache_beam/runners/common.py", line 475, in apache_beam.runners.common.DoFnInvoker.invoke_finish_bundle File "apache_beam/runners/common.py", line 481, in apache_beam.runners.common.DoFnInvoker.invoke_finish_bundle File "/usr/local/lib/python3.7/site-packages/apache_beam/io/gcp/bigtableio.py", line 187, in finish_bundle self.batcher.flush() File "/usr/local/lib/python3.7/site-packages/apache_beam/io/gcp/bigtableio.py", line 88, in flush status.code))) Exception: Failed to write a batch of 12 records due to 'not_found' During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py", line 267, in _execute response = task() File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py", line 340, in lambda: self.create_worker().do_instruction(request), request) File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py", line 581, in do_instruction getattr(request, request_type), request.instruction_id) File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py", line 618, in process_bundle bundle_processor.process_bundle(instruction_id)) File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1001, in process_bundle op.finish() File "apache_beam/runners/worker/operations.py", line 736, in apache_beam.runners.worker.operations.DoOperation.finish File "apache_beam/runners/worker/operations.py", line 738, in apache_beam.runners.worker.operations.DoOperation.finish File "apache_beam/runners/worker/operations.py", line 739, in apache_beam.runners.worker.operations.DoOperation.finish File "apache_beam/runners/common.py", line 1253, in apache_beam.runners.common.DoFnRunner.finish File "apache_beam/runners/common.py", line 1234, in apache_beam.runners.common.DoFnRunner._invoke_bundle_method File "apache_beam/runners/common.py", line 1281, in apache_beam.runners.common.DoFnRunner._reraise_augmented File "apache_beam/runners/common.py", line 1232, in apache_beam.runners.common.DoFnRunner._invoke_bundle_method File "apache_beam/runners/common.py", line 475, in apache_beam.runners.common.DoFnInvoker.invoke_finish_bundle File "apache_beam/runners/common.py", line 481, in apache_beam.runners.common.DoFnInvoker.invoke_finish_bundle File "/usr/local/lib/python3.7/site-packages/apache_beam/io/gcp/bigtableio.py", line 187, in finish_bundle self.batcher.flush() File "/usr/local/lib/python3.7/site-packages/apache_beam/io/gcp/bigtableio.py", line 88, in flush status.code))) Exception: Failed to write a batch of 12 records due to 'not_found' [while running 'WriteToBigTable/ParDo(_BigTableWriteFn)-ptransform-43']

排查建议 #

• 确认资源存在性：检查config.ini中配置的Cloud BigTable实例ID、表ID、列族ID是否完全正确，确认对应资源已在GCP控制台创建完成。
• 权限验证：确保Dataflow使用的服务账号拥有bigtable.tables.write权限，同时需具备访问Cloud BigTable实例的权限，可通过GCP IAM控制台检查服务账号权限配置。
• 依赖版本兼容性：当前使用的google-cloud-bigtable==1.7.0和apache-beam[gcp]==2.39.0版本可能存在兼容性问题，建议升级至匹配的新版本（例如将apache-beam升级到2.40+，对应调整bigtable客户端版本）。
• 列族存在性检查：代码中未验证列族是否存在，可取消注释代码中被注释的客户端初始化代码，添加列族存在性校验逻辑，避免因列族不存在导致写入失败。
• 网络连通性：确认Dataflow工作节点与Cloud BigTable实例处于同一VPC网络，或已配置正确的VPC peering/私网访问规则，避免因网络问题导致无法找到资源。

内容的提问来源于stack exchange，提问作者Pratibha