data-upload-java-cli

Domo 数据上传指南

首选策略 — 优先使用 JSON 连接器，CLI 作为回退： 当数据源是 REST/JSON API 时，始终先尝试 Domo JSON 无代码连接器（domo-json-no-code-connector skill）。它会在 Domo 内创建一个实时、可重复运行的连接器流，因此可以直接从 UI 刷新数据，而无需重新运行脚本。 仅在以下情况才回退到 Java CLI：

• JSON 连接器失败（认证问题、不支持的分页、解析错误），或
• 数据已经是 CSV 形式（本地文件、数据库导出等），且没有可连接的源 API。

概述

本指南介绍使用 domoutil.jar 将 CSV 数据上传到 Domo 的 Java CLI 回退方案。CLI 会自动处理 schema 推导、数据集创建、分块上传和索引。

JSON 连接器优先 / CLI 回退模式

当从 REST/JSON API 拉取数据到 Domo 时，在代码中使用这种双路径策略：

def ingest_object(obj_type, cfg):
    # Path 1 — JSON No-Code Connector (creates a live stream in Domo)
    ok, dataset_id = try_json_connector(obj_type, cfg)

    # Path 2 — CLI fallback (fetch → CSV → upload)
    if not ok:
        ok, dataset_id = try_cli_fallback(obj_type, cfg, work_dir)

路径 1（JSON 连接器） 步骤：

1. POST /api/data/v1/accounts — 创建一个 json5 账户用于认证
2. POST /api/data/v1/streams — 配置 URL、请求头、解析、分页
3. POST /api/data/v1/streams/{id}/executions — 触发一次运行
4. 轮询 GET /api/data/v1/streams/{id}/executions/{execId}，直到 currentState 为 "SUCCESS" 或 "ERROR"
5. 如果为 "ERROR" 或任一步骤出现 HTTP 错误 → 转入路径 2

路径 2（CLI 回退） 步骤：

1. 使用 requests 从源 API 拉取所有分页
2. 将记录写入 CSV 文件
3. derive-schema → 构建 schema JSON
4. create-dataset → 获取数据集 UUID
5. upload-dataset -h → 上传 CSV

何时直接跳到 CLI：

• 数据已经是 CSV（没有源 API）
• 源 API 不是 JSON/REST（SOAP、二进制等）
• 你已确认 JSON 连接器无法处理源的分页

JSON 连接器的认证说明： 使用 credentialsType: "fields" 搭配 authentication: "none"，并直接在 jsonSelection.httpsHeaders 中注入 Authorization 请求头。不要使用 authType — 它对 json5 无效，会返回 400。

Java CLI（回退路径）

• CLI 位置： /Users/elliottleonard/Documents/Cursor/CLI/domoutil.jar
• 运行命令： java -jar /Users/elliottleonard/Documents/Cursor/CLI/domoutil.jar

前置条件

• 已安装 Java
• 一个 Domo 访问令牌（格式：DDCI...）
• 一个 Domo 实例 URL（例如 instance.domo.com）
• 要上传的 CSV 文件（符合 RFC 4180）

快速参考

连接

connect -server <instance>.domo.com -token <API_TOKEN>

使用 whoami 验证。

完整工作流

步骤 1：从 CSV 推导 Schema

derive-schema -d /path/to/data.csv -r 500

• -d：CSV 文件的路径
• -r：用于类型检测的采样行数（500 是不错的默认值）
• 返回列名和检测到的类型：STRING、LONG、DOUBLE、DATETIME 等。

输出格式：

Schema:
  column_name1         STRING
  column_name2         LONG
  column_name3         DATETIME

步骤 2：创建 Schema JSON 文件

将 derive-schema 的输出转换为 JSON schema 文件：

{
  "columns": [
    {
      "name": "column_name1",
      "type": "STRING",
      "metadata": null,
      "upsertKey": false
    },
    {
      "name": "column_name2",
      "type": "LONG",
      "metadata": null,
      "upsertKey": false
    },
    {
      "name": "column_name3",
      "type": "DATETIME",
      "metadata": null,
      "upsertKey": false
    }
  ],
  "objects": []
}

有效的列类型： STRING、LONG、DOUBLE、DECIMAL、DATETIME、DATE

步骤 3：在 Domo 中创建数据集

create-dataset -n "<Dataset Name>" -t "<type>" -s /path/to/schema.json

• -n：数据集名称（在 Domo 中显示）
• -t：数据集类型（使用 "domo-cli" 作为通用类型）
• -s：schema JSON 文件的路径

输出：

Created DataSet: <dataset-uuid>

保存此 UUID — 上传步骤需要用到它。

步骤 4：上传 CSV 数据

upload-dataset -i <dataset-uuid> -f /path/to/data.csv -h

• -i / --id：创建步骤得到的数据集 UUID
• -f / --data：CSV 文件的路径
• -h / --headers：当 CSV 有表头行时必需（跳过第一行）

成功时的输出：

Started upload for DataSet <uuid>.
Finished upload for DataSet <uuid>. Bytes sent X.
Started indexing for dataset <uuid>
Finished indexing for dataset <uuid> with status SUCCESS
Data uploaded successfully

上传选项

Flag	说明
-i <ID>	数据集 UUID（必需）
-f <FILE>	CSV 文件路径
-h	CSV 有表头行
-a / --append	追加到现有数据，而不是替换
-d <DIR>	上传目录中的所有 CSV（文件不应包含表头）
-c	文件为 gzip 压缩（仅与 -d 一起使用）
-p <TAG>	分区标签（仅与 --append 一起使用）
-m <N>	最大上传线程数
-x	上传后跳过索引

脚本化 / 非交互模式

通过 stdin 管道传入命令以实现自动化：

echo -e "connect -server instance.domo.com -token YOUR_TOKEN\nupload-dataset -i <uuid> -f data.csv -h\nquit" \
  | java -jar /Users/elliottleonard/Documents/Cursor/CLI/domoutil.jar

务必以 quit 结尾，以确保干净退出。

批量上传：多个 CSV

Python 脚本模式

这是一次性上传多个 CSV 文件的推荐方法。该脚本会：

1. 为每个 CSV 推导 schema
2. 创建 schema JSON 文件
3. 在 Domo 中创建数据集
4. 上传 CSV 数据

import subprocess, json, os, re

csv_dir = "/path/to/csv/directory"
cli = "/Users/elliottleonard/Documents/Cursor/CLI/domoutil.jar"
server = "instance.domo.com"
token = "DDCI..."
schema_dir = "/tmp/domo_schemas"
os.makedirs(schema_dir, exist_ok=True)

csvs = sorted([f for f in os.listdir(csv_dir) if f.endswith('.csv')])

# --- Phase 1: Derive schemas and save as JSON ---
for csv_file in csvs:
    csv_path = os.path.join(csv_dir, csv_file)
    cmds = f"connect -server {server} -token {token}\nderive-schema -d {csv_path} -r 500\nquit"
    result = subprocess.run(['java', '-jar', cli], input=cmds, capture_output=True, text=True, timeout=60)

    columns = []
    in_schema = False
    for line in result.stdout.split('\n'):
        if 'Schema:' in line:
            in_schema = True
            continue
        if in_schema and line.startswith('  ') and line.strip():
            parts = line.split()
            if len(parts) >= 2:
                col_type = parts[-1]
                col_name = ' '.join(parts[:-1]).strip()
                columns.append({"name": col_name, "type": col_type, "metadata": None, "upsertKey": False})
        elif in_schema and line.startswith('>'):
            in_schema = False

    schema = {"columns": columns, "objects": []}
    schema_path = os.path.join(schema_dir, csv_file.replace('.csv', '.json'))
    with open(schema_path, 'w') as f:
        json.dump(schema, f, indent=2)

# --- Phase 2: Create datasets ---
cmds = [f"connect -server {server} -token {token}"]
for csv_file in csvs:
    schema_path = os.path.join(schema_dir, csv_file.replace('.csv', '.json'))
    # Generate a friendly name from the filename
    friendly_name = csv_file.replace('.csv', '').split('.')[-1].replace('_', ' ').title()
    cmds.append(f'create-dataset -n "{friendly_name}" -t "domo-cli" -s {schema_path}')
cmds.append("quit")

result = subprocess.run(['java', '-jar', cli], input='\n'.join(cmds), capture_output=True, text=True, timeout=120)

# Parse dataset IDs from output
dataset_ids = []
for line in result.stdout.split('\n'):
    match = re.search(r'Created DataSet: ([0-9a-f-]{36})', line)
    if match:
        dataset_ids.append(match.group(1))

# --- Phase 3: Upload data ---
cmds = [f"connect -server {server} -token {token}"]
for csv_file, dataset_id in zip(csvs, dataset_ids):
    csv_path = os.path.join(csv_dir, csv_file)
    cmds.append(f'upload-dataset -i {dataset_id} -f {csv_path} -h')
cmds.append("quit")

result = subprocess.run(['java', '-jar', cli], input='\n'.join(cmds), capture_output=True, text=True, timeout=600)
print(result.stdout)

纯 Shell 模式

对于更简单的场景，可使用 shell 脚本：

CLI="/Users/elliottleonard/Documents/Cursor/CLI/domoutil.jar"
SERVER="instance.domo.com"
TOKEN="DDCI..."

# Create commands file
cat > /tmp/upload_cmds.txt << EOF
connect -server $SERVER -token $TOKEN
upload-dataset -i <dataset-id-1> -f /path/to/file1.csv -h
upload-dataset -i <dataset-id-2> -f /path/to/file2.csv -h
quit
EOF

cat /tmp/upload_cmds.txt | java -jar "$CLI"

替换现有数据集中的数据

要替换现有数据集中的所有数据（而非追加），请使用相同的 upload-dataset 命令，但不要加 --append：

upload-dataset -i <existing-dataset-uuid> -f /path/to/new_data.csv -h

这会执行完全替换——所有现有行都会被移除，并替换为新的 CSV 数据。

追加数据

要在不移除现有数据的情况下向现有数据集添加行：

upload-dataset -i <dataset-uuid> -f /path/to/new_rows.csv -h --append

常见陷阱

1. 缺少 -h 标志

如果你的 CSV 带有表头行，而你又忘了加 -h，表头行会被当作数据导入。对于带表头的 CSV，请始终使用 -h。

2. upload-dataset 需要已存在的数据集

upload-dataset 命令不会创建数据集。你必须先用 schema 文件执行 create-dataset，然后用返回的 UUID 执行 upload-dataset。在没有有效 --id 的情况下上传会导致：

Executing POST: https://.../datasources/null/uploads
An error occurred while uploading the data from file

3. Schema 列名必须与 CSV 表头匹配

schema JSON 中的列名必须与 CSV 表头名称完全一致（区分大小写）。使用 derive-schema 可自动检测它们。

4. create-dataset 必须使用 -t 标志

数据集类型（-t）是必填项。可使用 "domo-cli" 作为安全的默认值：

create-dataset -n "My Dataset" -t "domo-cli" -s schema.json

5. 大文件会自动拆分

超过约 10MB 的文件会自动拆分为多个上传部分。CLI 会透明地处理这一点。对于非常大的上传，可考虑使用 -m 来控制线程数。

6. derive-schema 第一列解析问题

derive-schema 的输出有时会将第一列名称与其前面的文本拼接在一起。请务必对照 CSV 表头核验第一列名称。

7. 目录上传不应包含表头

使用 -d（目录上传）时，CSV 文件不应包含表头行。此模式专为预拆分好的分区数据而设计。

验证上传

上传后，验证数据是否可访问：

query-data -id <dataset-uuid> -q "SELECT * FROM table LIMIT 10"

或检查数据集元数据：

get-dataset -id <dataset-uuid>
get-schema -id <dataset-uuid>