Oracle云环境中如何通过URL让SQL*Loader获取CSV并创建表？

一、单CSV文件通过带认证令牌URL直接导入 #

Oracle自治数据库（ADB）提供的DBMS_CLOUD包是当前最直接的方案，完全替代旧的SQL*Loader拖拽操作，支持带认证令牌的URL拉取文件并直接建表导入。

操作步骤： #

1. 创建认证凭证
   根据URL的认证方式配置凭证：

   • 如果令牌作为密码传递：

     BEGIN
       DBMS_CLOUD.CREATE_CREDENTIAL(
         credential_name => 'TOKEN_AUTH_CRED',
         username => 'placeholder', -- 令牌认证时用户名可填占位符
         password => 'your_authentication_token'
       );
     END;
     /
     

   • 如果是Bearer令牌需放在请求头：

     BEGIN
       DBMS_CLOUD.CREATE_CREDENTIAL(
         credential_name => 'BEARER_TOKEN_CRED',
         params => JSON_OBJECT('auth_header' VALUE 'Bearer your_authentication_token')
       );
     END;
     /
     

2. 直接建表并导入CSV
   让ADB自动推断列类型完成原样导入（如果有表头需跳过）：

   BEGIN
     DBMS_CLOUD.CREATE_TABLE(
       table_name => 'TARGET_TABLE_NAME',
       credential_name => 'TOKEN_AUTH_CRED',
       file_uri_list => 'https://your-csv-url-with-token',
       format => JSON_OBJECT('type' VALUE 'CSV', 'skipheaders' VALUE '1')
     );
   END;
   /
   

   若已有表结构，用COPY_DATA导入数据：

   BEGIN
     DBMS_CLOUD.COPY_DATA(
       table_name => 'TARGET_TABLE_NAME',
       credential_name => 'TOKEN_AUTH_CRED',
       file_uri_list => 'https://your-csv-url-with-token',
       format => JSON_OBJECT('type' VALUE 'CSV', 'skipheaders' VALUE '1')
     );
   END;
   /
   

二、定期增量获取并导入的自动化流程 #

结合ADB自带工具实现全自动化，无需外部脚本或调度器：

1. 准备工作：维护导入日志表 #

先创建一张表记录已导入的文件，用于识别新文件：

CREATE TABLE IMPORTED_FILE_LOG(
  file_uri VARCHAR2(1000) PRIMARY KEY,
  import_time TIMESTAMP DEFAULT SYSTIMESTAMP
);

可选创建错误日志表记录导入失败信息：

CREATE TABLE IMPORT_ERROR_LOG(
  file_uri VARCHAR2(1000),
  error_msg VARCHAR2(4000),
  error_time TIMESTAMP DEFAULT SYSTIMESTAMP
);

2. 增量导入PL/SQL逻辑 #

编写块实现拉取源列表、筛选未导入文件、自动建表导入：

DECLARE
  l_source_list CLOB;
  l_file_list JSON_ARRAY_T;
  l_current_file VARCHAR2(1000);
  l_existing_files TABLE OF VARCHAR2(1000);
BEGIN
  -- 拉取源列表内容（假设源列表为JSON格式）
  l_source_list := DBMS_CLOUD.GET_OBJECT(
    credential_name => 'TOKEN_AUTH_CRED',
    object_uri => 'https://your-source-list-url'
  );
  l_file_list := JSON_ARRAY_T(l_source_list);

  -- 获取已导入的文件列表
  SELECT file_uri BULK COLLECT INTO l_existing_files FROM IMPORTED_FILE_LOG;

  -- 遍历源列表，处理未导入文件
  FOR i IN 0..l_file_list.get_size()-1 LOOP
    l_current_file := l_file_list.get_string(i);
    IF l_current_file NOT IN (SELECT column_value FROM TABLE(l_existing_files)) THEN
      BEGIN
        -- 以文件名作为表名（处理特殊字符避免报错）
        DBMS_CLOUD.CREATE_TABLE(
          table_name => 'IMPORT_' || DBMS_ASSERT.SIMPLE_SQL_NAME(REPLACE(SUBSTR(l_current_file, INSTR(l_current_file, '/', -1)+1), '.csv', '')),
          credential_name => 'TOKEN_AUTH_CRED',
          file_uri_list => l_current_file,
          format => JSON_OBJECT('type' VALUE 'CSV', 'skipheaders' VALUE '1')
        );
        -- 记录已导入文件
        INSERT INTO IMPORTED_FILE_LOG(file_uri) VALUES(l_current_file);
      EXCEPTION
        WHEN OTHERS THEN
          -- 记录错误信息
          INSERT INTO IMPORT_ERROR_LOG(file_uri, error_msg) VALUES(l_current_file, SQLERRM);
      END;
    END IF;
  END LOOP;
  COMMIT;
END;
/

3. 创建定时任务自动执行 #

用DBMS_SCHEDULER创建定时任务，比如每天凌晨2点执行：

BEGIN
  DBMS_SCHEDULER.CREATE_JOB(
    job_name => 'INCREMENTAL_CSV_IMPORT_JOB',
    job_type => 'PLSQL_BLOCK',
    job_action => 'BEGIN -- 此处粘贴上述增量导入的PL/SQL代码 END;',
    start_date => SYSTIMESTAMP,
    repeat_interval => 'FREQ=DAILY; BYHOUR=2; BYMINUTE=0; BYSECOND=0;',
    enabled => TRUE,
    comments => '每日增量导入源列表中的新CSV文件'
  );
END;
/

关键注意事项 #

• 令牌过期：若令牌有有效期，可在PL/SQL逻辑中加入令牌刷新逻辑（调用令牌获取接口更新凭证），或手动定期更新DBMS_CLOUD凭证。
• 权限：学术云ADB默认已授权DBMS_CLOUD相关权限，若出现权限错误联系管理员调整。
• 表名规范：用DBMS_ASSERT.SIMPLE_SQL_NAME处理文件名，避免特殊字符导致的SQL语法错误。

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