PyTigerGraph加载S3 Parquet到TigerGraph返回None问题排查

云原生阿岚

2026-8-29

问题背景

现有可正常运行的GSQL脚本，可直接从S3读取Parquet格式的顶点、边数据加载到TigerGraph的test_graph图中，脚本内容如下：

use graph test_graph
drop job test_job
drop data_source deep_disco_test

create data_source S3 deep_disco_test for graph test_graph

set my_source = "/home/ubuntu/s3.config"

CREATE LOADING JOB test_job for GRAPH test_graph {
       DEFINE FILENAME vertex1= "$deep_disco_test:{\"file.uris\":\"s3://signals-output/db_to_tg/identities/test_identities.parquet/\",\"file.reader.type\": \"parquet\",\"file.regexp\":\".parquet\"}";
       DEFINE FILENAME edge1=   "$deep_disco_test:{\"file.uris\":\"s3://signals-output/db_to_tg/relations/test_relations.parquet/\",\"file.reader.type\": \"parquet\",\"file.regexp\":\".parquet\"}";

       LOAD vertex1 to VERTEX identity VALUES($"dd_id",$"names",$"identity_type") USING JSON_FILE = "true";
       LOAD edge1 to EDGE relation VALUES($"src",$"dst",$"edge_type") USING JSON_FILE = "true";

}

run loading job test_job using EOF = "true"

需求是在Databricks环境中，通过PyTigerGraph的conn类调用conn.uploadFile()方法实现同等数据加载效果，操作步骤如下：

拼接GSQL语句字符串，创建顶点、边对应的两个加载作业，代码如下：

load_job = f'''

use graph {graph}
drop job {test_load_vertices}
drop job {test_load_edges}
drop data_source {data_source}
create data_source S3 {data_source} for graph {graph}

set {data_source} = "/home/ubuntu/s3.config"

CREATE LOADING JOB {test_load_vertices} FOR GRAPH {graph} {{
       DEFINE FILENAME MyDataSource;
       LOAD MyDataSource to VERTEX identity VALUES($"dd_id",$"names",$"identity_type") USING JSON_FILE = "true";}}
                                       
CREATE LOADING JOB {test_load_edges} FOR GRAPH {graph} {{
       DEFINE FILENAME MyDataSource;
       LOAD MyDataSource to EDGE relation VALUES($"src", $"dst", $"edge_type") USING JSON_FILE = "true";}}
'''

调用conn.gsql(load_job)加载上述作业，接口返回结果显示作业、数据源均创建成功：

"Using graph 'dd_subgraph_test'
Successfully dropped jobs on the graph 'dd_subgraph_test': [test_load_vertices_job].
Successfully dropped jobs on the graph 'dd_subgraph_test': [test_load_edges_job].
Successfully dropped data sources: [deep_disco_test].
Successfully created data sources: [deep_disco_test].
Data source 'deep_disco_test' has been updated.
Successfully created loading jobs: [test_load_vertices_job].
Successfully created loading jobs: [test_load_edges_job]."

测试运行顶点加载作业，调用代码如下：

print(conn.uploadFile(
  filePath="s3://signals-output/db_to_tg/identities/test_identities.parquet", 
  fileTag='MyDataSource', 
  jobName="test_load_vertices_job"))

执行后方法打印返回值为None，GraphStudio界面中无任何数据加载动作与记录。

问题根因

操作存在三个核心错误，直接导致加载失效：

conn.uploadFile()的逻辑是上传调用方本地文件系统的文件到TigerGraph执行加载，当前运行环境是Databricks，传入的filePath是S3路径，不是Databricks集群本地磁盘的真实文件路径，方法找不到对应文件直接静默失败，返回None，根本不会触发加载作业运行。
创建的两个加载作业里的MyDataSource变量没有配置Parquet读取规则，和原本可正常运行的GSQL脚本的FILENAME定义逻辑不一致：原脚本明确给文件变量绑定了S3数据源、Parquet文件读取器、parquet后缀匹配规则，当前定义里只有一个空的文件名占位符，就算文件上传成功也无法正确解析Parquet格式。
创建的S3数据源完全没有在加载作业里被引用，属于冗余配置，uploadFile走的是HTTP本地文件上传链路，和定义的S3数据源没有任何关联，根本不会用到S3的鉴权配置去读取S3文件。

修复方案

根据实际场景二选一即可：

推荐方案（和原GSQL脚本逻辑一致，性能最优）：放弃用uploadFile方法，直接让TigerGraph直连S3读取文件，不需要把文件中转到Databricks本地。把加载作业的FILENAME定义和原脚本对齐，创建完作业后直接通过GSQL语句触发运行即可，示例代码：

load_job = f'''
use graph {graph}
drop job if exists test_load_job
drop data_source if exists {data_source}
create data_source S3 {data_source} for graph {graph}
set {data_source} = "/home/ubuntu/s3.config"

CREATE LOADING JOB test_load_job FOR GRAPH {graph} {{
       DEFINE FILENAME vertex_file= "${data_source}:{{\\"file.uris\\":\\"s3://signals-output/db_to_tg/identities/test_identities.parquet/\\",\\"file.reader.type\\": \\"parquet\\",\\"file.regexp\\":\\".parquet\\"}}";
       DEFINE FILENAME edge_file= "${data_source}:{{\\"file.uris\\":\\"s3://signals-output/db_to_tg/relations/test_relations.parquet/\\",\\"file.reader.type\\": \\"parquet\\",\\"file.regexp\\":\\".parquet\\"}}";

       LOAD vertex_file to VERTEX identity VALUES($"dd_id",$"names",$"identity_type") USING JSON_FILE = "true";
       LOAD edge_file to EDGE relation VALUES($"src",$"dst",$"edge_type") USING JSON_FILE = "true";
}}
'''
# 创建加载作业
conn.gsql(load_job)
# 触发作业运行
run_res = conn.gsql(f"use graph {graph} run loading job test_load_job using EOF = 'true'")
print(run_res)

如果一定要用uploadFile方法：首先需要把S3上的Parquet文件下载到Databricks集群的本地临时磁盘路径，其次修改加载作业定义，给MyDataSource加上Parquet格式解析配置，删除无用的S3数据源配置，最后调用uploadFile时传入Databricks本地的文件路径，而不是S3路径。注意这种方案会多一层文件下载、HTTP上传的开销，数据量大时性能远低于TigerGraph直读S3的方案。

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