如何在GCP上运行自定义Python脚本实现ML模型批量预测？

针对你的需求——处理500万行CSV、自定义预处理、多Worker分布式执行、集成已训练的AI Platform模型——这里有几个最适合的GCP方案，每个都能满足你用自定义Python脚本完成任务的要求：

方案1：使用GCP Dataflow（Apache Beam）——推荐用于大规模ETL+预测 #

Dataflow是GCP托管的Apache Beam服务，天生为分布式数据处理设计，完美适配百万级数据量，支持自动扩缩容Worker，而且能轻松集成GCP的各种服务（包括AI Platform模型）。

步骤： #

1. 准备依赖：安装Beam和GCP相关库

pip install apache-beam[gcp] google-cloud-aiplatform

2. 编写自定义Beam管道脚本（示例）：
   这个脚本会读取GCS上的CSV，做预处理，调用AI Platform的模型预测，然后把结果写回GCS/BigQuery。

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions
from google.cloud import aiplatform

def preprocess_row(row):
    # 你的自定义预处理逻辑：转换指定列到模型要求的格式
    transformed_feature = transform_your_column(row['target_column'])
    return {**row, 'processed_feature': transformed_feature}

def predict_with_ai_platform(row):
    # 初始化AI Platform预测客户端
    client = aiplatform.PredictionServiceClient()
    endpoint = client.endpoint_path(
        project="your-project-id",
        location="us-central1",
        endpoint="your-model-endpoint-id"
    )
    # 构造预测请求
    instances = [row['processed_feature']]
    response = client.predict(endpoint=endpoint, instances=instances)
    # 把预测结果加入行数据
    return {**row, 'prediction': response.predictions[0]}

def run():
    options = PipelineOptions()
    google_cloud_options = options.view_as(GoogleCloudOptions)
    google_cloud_options.project = 'your-project-id'
    google_cloud_options.job_name = 'csv-prediction-job'
    google_cloud_options.staging_location = 'gs://your-bucket/staging'
    google_cloud_options.temp_location = 'gs://your-bucket/temp'
    google_cloud_options.region = 'us-central1'
    # 设置Worker数量（可根据需求调整）
    options.view_as(beam.options.pipeline_options.WorkerOptions).num_workers = 5

    with beam.Pipeline(options=options) as p:
        (
            p
            | "Read CSV from GCS" >> beam.io.ReadFromText('gs://your-bucket/input.csv', skip_header_lines=1)
            | "Parse CSV to Dict" >> beam.Map(lambda line: dict(zip(['col1', 'col2', 'target_column'], line.split(','))))
            | "Preprocess Data" >> beam.Map(preprocess_row)
            | "Run Prediction" >> beam.Map(predict_with_ai_platform)
            | "Write Results to GCS" >> beam.io.WriteToText('gs://your-bucket/output.csv', header='col1,col2,target_column,processed_feature,prediction')
        )

if __name__ == '__main__':
    run()

3. 提交Dataflow任务：

python your_script.py --runner DataflowRunner

方案2：使用GCP Dataproc（Spark集群）——适合熟悉Spark的场景 #

如果你熟悉PySpark，Dataproc可以快速创建Spark集群，分布式处理CSV，并行执行预处理和预测，效率极高。

步骤： #

1. 创建Dataproc集群（通过GCP控制台或gcloud命令）：

gcloud dataproc clusters create your-cluster-name \
    --region us-central1 \
    --num-workers 5 \
    --worker-machine-type n1-standard-4 \
    --initialization-actions gs://dataproc-initialization-actions/python/pip-install.sh \
    --metadata 'PIP_PACKAGES=google-cloud-aiplatform pandas'

2. 编写PySpark脚本（示例）：

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from google.cloud import aiplatform

def init_prediction_client():
    # 每个Worker初始化一次客户端
    client = aiplatform.PredictionServiceClient()
    endpoint = client.endpoint_path(
        project="your-project-id",
        location="us-central1",
        endpoint="your-model-endpoint-id"
    )
    return client, endpoint

def preprocess_and_predict(row):
    # 预处理逻辑
    transformed_feature = transform_your_column(row.target_column)
    # 调用模型预测
    client, endpoint = init_prediction_client()
    instances = [transformed_feature]
    response = client.predict(endpoint=endpoint, instances=instances)
    return row.col1, row.col2, row.target_column, transformed_feature, response.predictions[0]

if __name__ == '__main__':
    spark = SparkSession.builder.appName("CSV Prediction").getOrCreate()
    # 读取CSV（支持本地或GCS路径）
    df = spark.read.csv("gs://your-bucket/input.csv", header=True, inferSchema=True)
    # 注册UDF并执行
    predict_udf = udf(preprocess_and_predict, "string,string,string,string,float")
    result_df = df.select(predict_udf(df['col1'], df['col2'], df['target_column']).alias('result'))
    # 拆分结果列并保存
    result_df = result_df.select(
        result_df.result._1.alias('col1'),
        result_df.result._2.alias('col2'),
        result_df.result._3.alias('target_column'),
        result_df.result._4.alias('processed_feature'),
        result_df.result._5.alias('prediction')
    )
    # 保存到GCS或BigQuery
    result_df.write.csv("gs://your-bucket/output", header=True)
    spark.stop()

3. 提交PySpark任务到Dataproc：

gcloud dataproc jobs submit pyspark your_spark_script.py \
    --cluster your-cluster-name \
    --region us-central1 \
    --jars gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar

方案3：使用GCP Cloud Run Jobs + 任务拆分——轻量分布式方案 #

如果数据可以拆分多个小文件，你可以用Cloud Run Jobs创建多个并行任务，每个任务处理一部分CSV，最后合并结果。不过这个方案更适合中等规模数据，500万行也可以尝试：

1. 拆分CSV为多个小文件（用gsutil或本地脚本）：

gsutil cat gs://your-bucket/input.csv | split -l 100000 -d - "part_" && gsutil cp part_* gs://your-bucket/split_input/

2. 编写Cloud Run Job脚本：处理单个拆分文件，预处理+预测，保存结果。

3. 批量提交Cloud Run Jobs：用脚本循环提交每个拆分文件的处理任务，最后合并所有结果文件。

关键注意事项： #

• 模型访问：如果你的模型是AI Platform上的端点，确保Worker有足够的权限（给服务账号添加aiplatform.predictor角色）。
• 性能优化：对于Dataflow/Dataproc，调整Worker数量和机器类型可以提升处理速度；预处理逻辑尽量用Vectorized操作（比如Pandas的apply不如矢量化快）。
• 内存管理：500万行数据全放内存不现实，所以所有方案都是流式/分布式处理，避免加载全量数据到单节点内存。

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