Databricks使用COPY INTO导入CSV时字符串转整数失败问题

SQL解决方案 #

方法1：通过临时视图转换字段类型 #

先基于上传的CSV文件创建带类型转换的临时视图，再将数据导入目标表：

-- 创建临时视图，显式转换concept_id为整数
CREATE OR REPLACE TEMP VIEW temp_concept_data AS
SELECT
  TRY_CAST(concept_id AS INT) AS concept_id,
  -- 其他字段按需保留或转换
  column2,
  column3
FROM csv.`dbfs:/path/to/your/uploaded/file.csv`;

-- 将临时视图数据插入目标表
COPY INTO your_target_table
FROM temp_concept_data
FILEFORMAT = CSV
COPY_OPTIONS ('mergeSchema' = 'true');

用TRY_CAST替代CAST可避免个别行转换失败导致任务中断，转换失败的行该字段会返回NULL，方便后续排查问题。

方法2：在COPY INTO语句中直接转换字段 #

无需创建临时视图，直接在COPY INTO的子查询里处理字段转换：

COPY INTO your_target_table
(concept_id, column2, column3)
FROM (
  SELECT
    TRY_CAST(_c0 AS INT) AS concept_id,
    _c1 AS column2,
    _c2 AS column3
  FROM csv.`dbfs:/path/to/your/uploaded/file.csv`
)
FILEFORMAT = CSV;

注：_c0、_c1是CSV的列序号（无表头时用），如果CSV带表头，直接用对应列名替换即可。

本地程序化调用Python方案 #

如果已有可用的Python处理脚本，可通过以下两种方式从本地触发执行：

方式1：使用Databricks CLI提交作业 #

1. 本地配置CLI：执行databricks configure，输入Workspace的URL和个人访问令牌（PAT）。
2. 创建作业配置文件job_config.json：

{
  "name": "CSV数据导入处理",
  "tasks": [
    {
      "task_key": "process_csv",
      "new_cluster": {
        "spark_version": "13.3.x-scala2.12",
        "node_type_id": "Standard_DS3_v2",
        "num_workers": 1
      },
      "spark_python_task": {
        "python_file": "/dbfs/path/to/your/python_script.py"
      }
    }
  ]
}

3. 提交作业：

databricks jobs submit --json-file job_config.json

方式2：使用Databricks Python SDK #

1. 安装SDK：pip install databricks-sdk
2. 编写本地触发脚本：

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import Task, SparkPythonTask, NewCluster

# 初始化客户端，自动读取本地配置的PAT和Workspace URL
w = WorkspaceClient()

# 定义作业任务
task = Task(
  task_key="process_csv_task",
  new_cluster=NewCluster(
    spark_version="13.3.x-scala2.12",
    node_type_id="Standard_DS3_v2",
    num_workers=1
  ),
  spark_python_task=SparkPythonTask(
    python_file="/dbfs/path/to/your/python_script.py"
  )
)

# 提交作业并返回Run ID
job_run = w.jobs.submit(run_name="CSV数据处理", tasks=[task])
print(f"作业已提交，Run ID: {job_run.run_id}")

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