data-import

数据导入

概述

此技能涵盖使用 Import Sets 和 Transform Maps 在 ServiceNow 中完成数据导入的整个生命周期：

• 了解 Import Set 架构
• 配置数据源（文件、JDBC、LDAP、REST）
• 创建和配置 Transform Maps
• 字段映射策略和转换
• 转换脚本（onBefore、onAfter、onStart、onComplete）
• 用于匹配和去重的 Coalesce 字段
• 错误处理和回滚策略
• 计划导入和自动化
• 大型数据集的性能优化

使用场景： 将外部数据导入 ServiceNow、执行 ETL 操作、在系统之间迁移数据，或设置定期数据同步时。

适用对象： 开发人员、管理员、集成专家和数据迁移团队。

前置条件

• 角色： import_admin、import_transformer 或 admin
• 访问权限： 目标表、Import Set 表和数据源配置
• 知识要求： ServiceNow 数据模型、GlideRecord API、表关系
• 相关技能：
  • admin/generic-crud-operations - 基本 CRUD 操作
  • admin/batch-operations - 批量数据处理

Import Set 架构

关键组件

┌─────────────────┐    ┌──────────────────┐    ┌─────────────────┐
│   Data Source   │───>│   Import Set     │───>│  Transform Map  │
│  (File/JDBC/    │    │    (Staging)     │    │   (Mapping)     │
│   LDAP/REST)    │    │                  │    │                 │
└─────────────────┘    └──────────────────┘    └─────────────────┘
                                                       │
                                                       v
                                               ┌─────────────────┐
                                               │  Target Table   │
                                               │   (Production)  │
                                               └─────────────────┘

核心表

表	用途
sys_import_set	Import Set 表头记录
sys_import_set_row	导入数据的暂存表
sys_transform_map	Transform Map 定义
sys_transform_entry	字段映射条目
sys_transform_script	转换脚本（onBefore 等）
sys_data_source	数据源配置

导入状态

状态	值	说明
已加载	loaded	数据已加载到暂存区
已转换	transformed	转换成功
错误	error	转换失败
已忽略	ignored	被转换逻辑跳过

流程

阶段 1：数据源配置

步骤 1.1：创建文件数据源

用于 CSV、Excel 或 XML 文件导入。

使用 MCP：

Tool: SN-Create-Record
Parameters:
  table_name: sys_data_source
  data:
    name: Employee Import - CSV
    type: File
    format: CSV
    header_row: 1
    sheet_number: 1
    import_set_table_name: u_employee_import
    active: true

响应：

{
  "sys_id": "abc123...",
  "name": "Employee Import - CSV",
  "type": "File",
  "import_set_table_name": "u_employee_import"
}

步骤 1.2：创建 JDBC 数据源

用于数据库连接（Oracle、MySQL、SQL Server）。

Tool: SN-Create-Record
Parameters:
  table_name: sys_data_source
  data:
    name: HR Database - JDBC
    type: JDBC
    connection_url: jdbc:mysql://hr-db.company.com:3306/hrms
    user: servicenow_reader
    password: [encrypted_password]
    import_set_table_name: u_hr_import
    query: |
      SELECT employee_id, first_name, last_name, email, department, hire_date
      FROM employees
      WHERE modified_date > ?
    active: true

JDBC 连接 URL 格式：

数据库	连接 URL
MySQL	jdbc:mysql://host:3306/database
Oracle	jdbc:oracle:thin:@host:1521:sid
SQL Server	jdbc:sqlserver://host:1433;databaseName=db
PostgreSQL	jdbc:postgresql://host:5432/database

步骤 1.3：创建 LDAP 数据源

用于导入 Active Directory 或 LDAP 目录。

Tool: SN-Create-Record
Parameters:
  table_name: sys_data_source
  data:
    name: Active Directory Users
    type: LDAP
    server_url: ldap://ad.company.com:389
    user: CN=ServiceNow,OU=Service Accounts,DC=company,DC=com
    password: [encrypted_password]
    import_set_table_name: u_ldap_user_import
    ldap_target: OU=Users,DC=company,DC=com
    ldap_filter: (&(objectClass=user)(!(userAccountControl:1.2.840.113556.1.4.803:=2)))
    active: true

步骤 1.4：创建 REST 数据源

用于 REST API 集成。

Tool: SN-Create-Record
Parameters:
  table_name: sys_data_source
  data:
    name: External API - REST
    type: REST (IntegrationHub)
    connection_url: https://api.external-system.com/v1/records
    http_method: GET
    authentication_type: basic
    user: api_user
    password: [encrypted_password]
    import_set_table_name: u_api_import
    format: JSON
    active: true

阶段 2：创建 Import Set 表

步骤 2.1：创建自定义 Import Set 表

创建一个暂存表，用于接收导入的数据。

使用 MCP：

Tool: SN-Create-Record
Parameters:
  table_name: sys_db_object
  data:
    name: u_employee_import
    label: Employee Import
    extends: sys_import_set_row
    create_access: true
    read_access: true
    update_access: true
    delete_access: true

步骤 2.2：向 Import Set 表添加列

定义与源数据结构匹配的列。

批量创建列：

Tool: SN-Batch-Create
Parameters:
  records:
    - table_name: sys_dictionary
      data:
        name: u_employee_import
        element: u_employee_id
        column_label: Employee ID
        internal_type: string
        max_length: 40
    - table_name: sys_dictionary
      data:
        name: u_employee_import
        element: u_first_name
        column_label: First Name
        internal_type: string
        max_length: 100
    - table_name: sys_dictionary
      data:
        name: u_employee_import
        element: u_last_name
        column_label: Last Name
        internal_type: string
        max_length: 100
    - table_name: sys_dictionary
      data:
        name: u_employee_import
        element: u_email
        column_label: Email
        internal_type: string
        max_length: 255
    - table_name: sys_dictionary
      data:
        name: u_employee_import
        element: u_department
        column_label: Department
        internal_type: string
        max_length: 100
    - table_name: sys_dictionary
      data:
        name: u_employee_import
        element: u_hire_date
        column_label: Hire Date
        internal_type: string
        max_length: 40

阶段 3：配置 Transform Map

步骤 3.1：创建 Transform Map

定义如何将暂存数据转换到目标表。

使用 MCP：

Tool: SN-Create-Record
Parameters:
  table_name: sys_transform_map
  data:
    name: Employee Import Transform
    source_table: u_employee_import
    target_table: sys_user
    active: true
    enforce_mandatory_fields: true
    run_business_rules: true
    run_script: true
    order: 100

Transform Map 选项：

选项	说明
enforce_mandatory_fields	缺少必填字段时失败
run_business_rules	在目标表上执行业务规则
run_script	运行转换脚本
copy_empty_fields	使用空值覆盖
order	执行顺序（数值越小越早）

步骤 3.2：创建字段映射

将源列映射到目标字段。

批量创建字段映射：

Tool: SN-Batch-Create
Parameters:
  records:
    - table_name: sys_transform_entry
      data:
        map: [transform_map_sys_id]
        source_field: u_employee_id
        target_field: employee_number
        coalesce: true
        order: 100
    - table_name: sys_transform_entry
      data:
        map: [transform_map_sys_id]
        source_field: u_first_name
        target_field: first_name
        order: 200
    - table_name: sys_transform_entry
      data:
        map: [transform_map_sys_id]
        source_field: u_last_name
        target_field: last_name
        order: 300
    - table_name: sys_transform_entry
      data:
        map: [transform_map_sys_id]
        source_field: u_email
        target_field: email
        coalesce: true
        order: 400
    - table_name: sys_transform_entry
      data:
        map: [transform_map_sys_id]
        source_field: u_department
        target_field: department
        reference_qual_mapping: true
        order: 500

第 3.3 步：字段映射类型

类型	用例	配置
直接	简单复制	从源字段复制到目标字段，不进行转换
映射	值转换	使用选项映射或脚本
引用	查找关联	设置 reference_qual_mapping: true
脚本	复杂逻辑	使用 source_script 字段
派生	计算所得	无源字段，仅使用脚本

脚本映射示例：

Tool: SN-Create-Record
Parameters:
  table_name: sys_transform_entry
  data:
    map: [transform_map_sys_id]
    source_field: u_status
    target_field: active
    use_source_script: true
    source_script: |
      // Convert status to boolean active flag
      answer = (source.u_status == 'Active' || source.u_status == 'A') ? 'true' : 'false';
    order: 600

带查找的引用映射：

Tool: SN-Create-Record
Parameters:
  table_name: sys_transform_entry
  data:
    map: [transform_map_sys_id]
    source_field: u_manager_email
    target_field: manager
    reference_qual_mapping: true
    reference_qual: email=[u_manager_email]
    order: 700

第 4 阶段：合并字段（匹配）

第 4.1 步：了解合并

合并字段用于确定转换应执行插入还是更新：

• 无匹配项： 插入新记录
• 单个匹配项： 更新现有记录
• 多个匹配项： 出错（除非另有配置）

第 4.2 步：配置合并字段

单个合并字段：

Tool: SN-Update-Record
Parameters:
  table_name: sys_transform_entry
  sys_id: [entry_sys_id]
  data:
    coalesce: true

多个合并字段（复合键）：

Tool: SN-Batch-Update
Parameters:
  updates:
    - table_name: sys_transform_entry
      sys_id: [employee_id_entry_sys_id]
      data:
        coalesce: true
    - table_name: sys_transform_entry
      sys_id: [company_entry_sys_id]
      data:
        coalesce: true

合并行为矩阵：

场景	行为
无合并字段	始终插入新记录
有合并字段，无匹配项	插入新记录
有合并字段，匹配一条记录	更新现有记录
有合并字段，匹配多条记录	出错（可配置）

第 4.3 步：处理多个匹配项

配置转换映射以处理多个匹配项。

Tool: SN-Update-Record
Parameters:
  table_name: sys_transform_map
  sys_id: [transform_map_sys_id]
  data:
    multi_coalesce_action: ignore

多重合并操作：

操作	行为
create	仍然创建新记录
ignore	跳过此行，并标记为已忽略
update_first	更新第一条匹配记录
reject	将此行标记为错误

第 5 阶段：转换脚本

第 5.1 步：脚本类型概览

脚本类型	执行时机	用例
onStart	转换开始前	初始化计数器、执行验证
onBefore	每行转换前	行级预处理
onAfter	每行转换后	后处理、处理关联记录
onComplete	转换结束后	汇总、通知
onChoiceCreate	创建选项时	自定义选项创建
onForeignInsert	插入引用时	处理缺失的引用

第 5.2 步：创建 onStart 脚本

在转换开始时运行一次。

Tool: SN-Create-Record
Parameters:
  table_name: sys_transform_script
  data:
    map: [transform_map_sys_id]
    script_type: onStart
    script: |
      // onStart: Initialize transform
      // Available: log, source (first row), map, import_set

      log.info('Starting employee import transform');
      log.info('Import Set: ' + import_set.number);
      log.info('Source table: ' + map.source_table);

      // Initialize counters in scratchpad
      var scratchpad = {};
      scratchpad.processed = 0;
      scratchpad.created = 0;
      scratchpad.updated = 0;
      scratchpad.errors = 0;
      scratchpad.startTime = new GlideDateTime();
    order: 100
    active: true

第 5.3 步：创建 onBefore 脚本

在每行转换前运行。

Tool: SN-Create-Record
Parameters:
  table_name: sys_transform_script
  data:
    map: [transform_map_sys_id]
    script_type: onBefore
    script: |
      // onBefore: Row-level preprocessing
      // Available: source, target, map, log, action, error, ignore
      // Set ignore=true to skip row, error=true to mark as error

      // Validate required fields
      if (!source.u_employee_id || source.u_employee_id.nil()) {
        error = true;
        error_message = 'Missing employee ID';
        return;
      }

      if (!source.u_email || source.u_email.nil()) {
        error = true;
        error_message = 'Missing email address';
        return;
      }

      // Validate email format
      var emailRegex = /^[^\s@]+@[^\s@]+\.[^\s@]+$/;
      if (!emailRegex.test(source.u_email.toString())) {
        error = true;
        error_message = 'Invalid email format: ' + source.u_email;
        return;
      }

      // Normalize data
      source.u_first_name = source.u_first_name.toString().trim();
      source.u_last_name = source.u_last_name.toString().trim();
      source.u_email = source.u_email.toString().toLowerCase().trim();

      // Generate username if not provided
      if (!source.u_user_name || source.u_user_name.nil()) {
        source.u_user_name = source.u_email.toString().split('@')[0];
      }

      // Conditional skip
      if (source.u_status == 'Terminated') {
        ignore = true;
        return;
      }

      scratchpad.processed++;
    order: 100
    active: true

第 5.4 步：创建 onAfter 脚本

在每行转换后运行。

Tool: SN-Create-Record
Parameters:
  table_name: sys_transform_script
  data:
    map: [transform_map_sys_id]
    script_type: onAfter
    script: |
      // onAfter: Post-processing
      // Available: source, target, map, log, action, error, scratchpad
      // action = 'insert', 'update', or 'ignore'

      if (action == 'insert') {
        scratchpad.created++;

        // Add user to default groups
        if (target.sys_id) {
          addUserToGroups(target.sys_id, source.u_department);
        }

        log.info('Created user: ' + target.user_name);

      } else if (action == 'update') {
        scratchpad.updated++;
        log.info('Updated user: ' + target.user_name);
      }

      // Create related records
      if (source.u_manager_email && !source.u_manager_email.nil()) {
        // Store for later processing
        scratchpad.managersToProcess = scratchpad.managersToProcess || [];
        scratchpad.managersToProcess.push({
          userId: target.sys_id.toString(),
          managerEmail: source.u_manager_email.toString()
        });
      }

      function addUserToGroups(userId, department) {
        var deptGroups = {
          'IT': ['IT Support', 'Service Desk'],
          'HR': ['HR Team'],
          'Finance': ['Finance Team']
        };

        var groups = deptGroups[department] || [];

        groups.forEach(function(groupName) {
          var group = new GlideRecord('sys_user_group');
          group.addQuery('name', groupName);
          group.query();

          if (group.next()) {
            var member = new GlideRecord('sys_user_grmember');
            member.addQuery('user', userId);
            member.addQuery('group', group.sys_id);
            member.query();

            if (!member.hasNext()) {
              member.initialize();
              member.user = userId;
              member.group = group.sys_id;
              member.insert();
            }
          }
        });
      }
    order: 100
    active: true

第 5.5 步：创建 onComplete 脚本

在所有行转换完成后运行一次。

Tool: SN-Create-Record
Parameters:
  table_name: sys_transform_script
  data:
    map: [transform_map_sys_id]
    script_type: onComplete
    script: |
      // onComplete: Finalize transform
      // Available: log, import_set, map, scratchpad

      var endTime = new GlideDateTime();
      var duration = GlideDateTime.subtract(scratchpad.startTime, endTime);
      var durationSec = duration.getNumericValue() / 1000;

      var summary = {
        processed: scratchpad.processed || 0,
        created: scratchpad.created || 0,
        updated: scratchpad.updated || 0,
        errors: scratchpad.errors || 0,
        duration: durationSec.toFixed(2) + ' seconds'
      };

      log.info('Transform Complete: ' + JSON.stringify(summary));

      // Process manager relationships (deferred to avoid reference issues)
      if (scratchpad.managersToProcess && scratchpad.managersToProcess.length > 0) {
        scratchpad.managersToProcess.forEach(function(item) {
          var manager = new GlideRecord('sys_user');
          manager.addQuery('email', item.managerEmail);
          manager.query();

          if (manager.next()) {
            var user = new GlideRecord('sys_user');
            if (user.get(item.userId)) {
              user.manager = manager.sys_id;
              user.update();
            }
          }
        });
        log.info('Processed ' + scratchpad.managersToProcess.length + ' manager relationships');
      }

      // Send notification if errors occurred
      if (summary.errors > 0) {
        gs.eventQueue('import.transform.errors', import_set, summary.errors, JSON.stringify(summary));
      }

      // Update import set with summary
      import_set.description = 'Summary: ' + JSON.stringify(summary);
      import_set.update();
    order: 100
    active: true

第 5.6 步：onForeignInsert 脚本

处理缺失的引用值。

Tool: SN-Create-Record
Parameters:
  table_name: sys_transform_script
  data:
    map: [transform_map_sys_id]
    script_type: onForeignInsert
    script: |
      // onForeignInsert: Handle missing references
      // Available: source, target_table, target_field, source_field, source_value, log
      // Return: sys_id of existing/created record, or ignore to skip

      // Handle department lookup - create if not exists
      if (target_table == 'cmn_department' && target_field == 'department') {
        var dept = new GlideRecord('cmn_department');
        dept.addQuery('name', source_value);
        dept.query();

        if (dept.next()) {
          return dept.sys_id;
        } else {
          // Create new department
          dept.initialize();
          dept.name = source_value;
          dept.primary_contact = ''; // Set to admin later
          var newId = dept.insert();
          log.info('Created department: ' + source_value);
          return newId;
        }
      }

      // Handle company lookup - ignore if not found
      if (target_table == 'core_company' && target_field == 'company') {
        log.warn('Company not found: ' + source_value);
        ignore = true;
        return;
      }

      // Default: skip the field
      ignore = true;
    order: 100
    active: true

第 6 阶段：错误处理

第 6.1 步：转换级错误处理

配置转换映射的错误行为。

Tool: SN-Update-Record
Parameters:
  table_name: sys_transform_map
  sys_id: [transform_map_sys_id]
  data:
    abort_on_error: false
    log_transform_messages: true

错误配置选项：

选项	描述
abort_on_error	遇到第一个错误时停止转换
log_transform_messages	写入详细日志
copy_empty_fields	包含空值

第 6.2 步：查询错误行

查找并分析失败的行。

使用 MCP：

Tool: SN-Query-Table
Parameters:
  table_name: sys_import_set_row
  query: sys_import_set=[import_set_sys_id]^sys_import_state=error
  fields: sys_id,sys_row_error,sys_import_state,sys_transform_map
  limit: 100

第 6.3 步：重试失败的行

修复问题后，重新处理错误行。

Tool: SN-Execute-Background-Script
Parameters:
  script: |
    // Retry failed import set rows
    var importSetId = '[import_set_sys_id]';
    var transformMapId = '[transform_map_sys_id]';

    var gr = new GlideRecord('sys_import_set_row');
    gr.addQuery('sys_import_set', importSetId);
    gr.addQuery('sys_import_state', 'error');
    gr.query();

    var retried = 0;
    var transformer = new GlideImportSetTransformer();

    while (gr.next()) {
      // Reset state
      gr.sys_import_state = 'pending';
      gr.sys_row_error = '';
      gr.update();

      // Retransform single row
      transformer.transformRow(gr, transformMapId);
      retried++;
    }

    gs.info('Retried ' + retried + ' rows');
  description: Retry failed import rows

第 6.4 步：全面错误日志记录

创建错误跟踪表并记录日志。

Tool: SN-Execute-Background-Script
Parameters:
  script: |
    // Enhanced error tracking for imports
    function logImportError(source, target, errorMsg, context) {
      var log = new GlideRecord('u_import_error_log');
      log.initialize();
      log.u_import_set = context.importSetId;
      log.u_transform_map = context.transformMapId;
      log.u_source_row = source.sys_id;
      log.u_source_data = JSON.stringify({
        employee_id: source.u_employee_id.toString(),
        email: source.u_email.toString(),
        name: source.u_first_name + ' ' + source.u_last_name
      });
      log.u_error_message = errorMsg;
      log.u_timestamp = new GlideDateTime();
      log.insert();

      return log.sys_id;
    }

    // Usage in onBefore script:
    // if (validationFailed) {
    //   logImportError(source, target, 'Validation failed: missing email', {
    //     importSetId: import_set.sys_id,
    //     transformMapId: map.sys_id
    //   });
    //   error = true;
    // }

    gs.info('Error logging function defined');
  description: Define import error logging function

第 7 阶段：计划导入

第 7.1 步：创建计划导入

设置定期重复的数据导入。

Tool: SN-Create-Record
Parameters:
  table_name: scheduled_import_set
  data:
    name: Daily Employee Sync
    data_source: [data_source_sys_id]
    transform_map: [transform_map_sys_id]
    run_type: daily
    run_time: "02:00:00"
    run_dayofweek: "*"
    active: true
    delete_on_success: true
    email_on_error: [email protected]

运行类型选项：

类型	描述	附加字段
on_demand	手动执行	无
daily	每天一次	run_time
weekly	每周一次	run_time、run_dayofweek
monthly	每月一次	run_time、run_dayofmonth
periodically	固定间隔	run_period（分钟）

第 7.2 步：创建导入集运行脚本

用于自定义计划或复杂工作流。

Tool: SN-Create-Record
Parameters:
  table_name: sysauto_script
  data:
    name: Employee Import - Custom Schedule
    script: |
      // Custom scheduled import with pre/post processing
      var startTime = new GlideDateTime();
      gs.info('Starting scheduled employee import');

      try {
        // Pre-import validation
        if (!validateSourceConnection()) {
          gs.error('Source connection validation failed - aborting import');
          return;
        }

        // Run import
        var dataSourceId = '[data_source_sys_id]';
        var transformMapId = '[transform_map_sys_id]';

        // Load data into import set
        var loader = new GlideImportSetLoader();
        var importSetId = loader.loadAttachment(dataSourceId);

        if (!importSetId) {
          gs.error('Failed to load import set');
          return;
        }

        // Transform data
        var transformer = new GlideImportSetTransformer();
        transformer.setImportSetID(importSetId);
        transformer.setTransformMapID(transformMapId);
        transformer.transform();

        // Post-import processing
        var stats = getImportStats(importSetId);
        gs.info('Import complete: ' + JSON.stringify(stats));

        // Send summary notification
        if (stats.errors > 0) {
          sendErrorNotification(stats);
        }

        // Cleanup old import sets (keep last 7 days)
        cleanupOldImportSets(7);

      } catch (e) {
        gs.error('Scheduled import failed: ' + e.message);
      }

      function validateSourceConnection() {
        // Add connection validation logic
        return true;
      }

      function getImportStats(importSetId) {
        var stats = { total: 0, inserted: 0, updated: 0, errors: 0, ignored: 0 };

        var ga = new GlideAggregate('sys_import_set_row');
        ga.addQuery('sys_import_set', importSetId);
        ga.addAggregate('COUNT');
        ga.groupBy('sys_import_state');
        ga.query();

        while (ga.next()) {
          var state = ga.sys_import_state.toString();
          var count = parseInt(ga.getAggregate('COUNT'));
          stats.total += count;

          if (state == 'inserted') stats.inserted = count;
          else if (state == 'updated') stats.updated = count;
          else if (state == 'error') stats.errors = count;
          else if (state == 'ignored') stats.ignored = count;
        }

        return stats;
      }

      function sendErrorNotification(stats) {
        gs.eventQueue('import.scheduled.errors', null, stats.errors, JSON.stringify(stats));
      }

      function cleanupOldImportSets(daysToKeep) {
        var gr = new GlideRecord('sys_import_set');
        gr.addQuery('sys_created_on', '<', gs.daysAgo(daysToKeep));
        gr.addQuery('state', 'processed');
        gr.query();

        var deleted = 0;
        while (gr.next()) {
          gr.deleteRecord();
          deleted++;
        }

        gs.info('Cleaned up ' + deleted + ' old import sets');
      }
    run_type: daily
    run_time: "02:00:00"
    active: true

第 8 阶段：性能优化

第 8.1 步：大型导入的最佳实践

导入集分块：

Tool: SN-Execute-Background-Script
Parameters:
  script: |
    // Process large imports in chunks
    var CONFIG = {
      dataSourceId: '[data_source_sys_id]',
      transformMapId: '[transform_map_sys_id]',
      chunkSize: 5000,
      pauseBetweenChunks: 5000  // milliseconds
    };

    var totalProcessed = 0;
    var chunkNumber = 0;

    while (true) {
      chunkNumber++;
      gs.info('Processing chunk ' + chunkNumber);

      // Load chunk
      var loader = new GlideImportSetLoader();
      loader.setLimit(CONFIG.chunkSize);
      loader.setOffset((chunkNumber - 1) * CONFIG.chunkSize);

      var importSetId = loader.loadAttachment(CONFIG.dataSourceId);

      if (!importSetId) {
        gs.info('No more data to process');
        break;
      }

      // Count rows loaded
      var ga = new GlideAggregate('sys_import_set_row');
      ga.addQuery('sys_import_set', importSetId);
      ga.addAggregate('COUNT');
      ga.query();

      var rowCount = 0;
      if (ga.next()) {
        rowCount = parseInt(ga.getAggregate('COUNT'));
      }

      if (rowCount == 0) {
        gs.info('Empty chunk - import complete');
        break;
      }

      // Transform chunk
      var transformer = new GlideImportSetTransformer();
      transformer.setImportSetID(importSetId);
      transformer.setTransformMapID(CONFIG.transformMapId);
      transformer.transform();

      totalProcessed += rowCount;
      gs.info('Chunk ' + chunkNumber + ' complete: ' + rowCount + ' rows (Total: ' + totalProcessed + ')');

      // Pause between chunks
      gs.sleep(CONFIG.pauseBetweenChunks);
    }

    gs.info('Import complete: ' + totalProcessed + ' total rows processed');
  description: Chunked large import processing

步骤 8.2：导入期间禁用业务规则

为大规模导入实现最佳性能。

Tool: SN-Update-Record
Parameters:
  table_name: sys_transform_map
  sys_id: [transform_map_sys_id]
  data:
    run_business_rules: false

警告： 只有在了解其影响后才禁用。导入后如有需要，请重新启用。

步骤 8.3：为合并字段创建索引

确保为合并字段创建索引，以提升性能。

Tool: SN-Create-Record
Parameters:
  table_name: sys_index
  data:
    table: sys_user
    field_list: employee_number,email
    unique: true
    active: true

步骤 8.4：性能监控脚本

Tool: SN-Execute-Background-Script
Parameters:
  script: |
    // Monitor import performance
    var importSetId = '[import_set_sys_id]';

    var importSet = new GlideRecord('sys_import_set');
    if (importSet.get(importSetId)) {
      var stats = {
        number: importSet.number.toString(),
        state: importSet.state.toString(),
        created: importSet.sys_created_on.toString(),
        updated: importSet.sys_updated_on.toString()
      };

      // Calculate duration
      var created = new GlideDateTime(importSet.sys_created_on);
      var updated = new GlideDateTime(importSet.sys_updated_on);
      var duration = GlideDateTime.subtract(created, updated);
      stats.duration_seconds = duration.getNumericValue() / 1000;

      // Get row counts by state
      var ga = new GlideAggregate('sys_import_set_row');
      ga.addQuery('sys_import_set', importSetId);
      ga.addAggregate('COUNT');
      ga.groupBy('sys_import_state');
      ga.query();

      stats.rows = {};
      while (ga.next()) {
        stats.rows[ga.sys_import_state.toString()] = parseInt(ga.getAggregate('COUNT'));
      }

      // Calculate throughput
      var totalRows = 0;
      for (var state in stats.rows) {
        totalRows += stats.rows[state];
      }
      stats.total_rows = totalRows;
      stats.rows_per_second = (totalRows / stats.duration_seconds).toFixed(2);

      gs.info('Import Performance:\n' + JSON.stringify(stats, null, 2));
    }
  description: Monitor import set performance

阶段 9：回滚和清理

步骤 9.1：回滚已插入的记录

撤销转换创建的记录。

Tool: SN-Execute-Background-Script
Parameters:
  script: |
    // Rollback inserted records from import
    var importSetId = '[import_set_sys_id]';
    var targetTable = 'sys_user';
    var DRY_RUN = true;  // Set to false to actually delete

    // Get inserted records
    var gr = new GlideRecord('sys_import_set_row');
    gr.addQuery('sys_import_set', importSetId);
    gr.addQuery('sys_import_state', 'inserted');
    gr.query();

    var toDelete = [];
    while (gr.next()) {
      toDelete.push(gr.sys_target_sys_id.toString());
    }

    gs.info('Found ' + toDelete.length + ' records to rollback');

    if (DRY_RUN) {
      gs.info('[DRY RUN] Would delete: ' + toDelete.join(', '));
    } else {
      var deleted = 0;
      toDelete.forEach(function(sysId) {
        var target = new GlideRecord(targetTable);
        if (target.get(sysId)) {
          target.deleteRecord();
          deleted++;
        }
      });
      gs.info('Deleted ' + deleted + ' records');

      // Update import set rows
      var updateGr = new GlideRecord('sys_import_set_row');
      updateGr.addQuery('sys_import_set', importSetId);
      updateGr.addQuery('sys_import_state', 'inserted');
      updateGr.query();

      while (updateGr.next()) {
        updateGr.sys_import_state = 'rolled_back';
        updateGr.update();
      }
    }
  description: Rollback imported records

步骤 9.2：清理导入集数据

删除旧的导入集和暂存数据。

Tool: SN-Execute-Background-Script
Parameters:
  script: |
    // Cleanup old import sets
    var DAYS_TO_KEEP = 30;
    var DRY_RUN = true;

    var cutoffDate = gs.daysAgo(DAYS_TO_KEEP);

    // Find old import sets
    var gr = new GlideRecord('sys_import_set');
    gr.addQuery('sys_created_on', '<', cutoffDate);
    gr.addQuery('state', 'processed');
    gr.query();

    var stats = { sets: 0, rows: 0 };

    while (gr.next()) {
      // Count rows
      var ga = new GlideAggregate('sys_import_set_row');
      ga.addQuery('sys_import_set', gr.sys_id);
      ga.addAggregate('COUNT');
      ga.query();

      var rowCount = 0;
      if (ga.next()) {
        rowCount = parseInt(ga.getAggregate('COUNT'));
      }

      if (DRY_RUN) {
        gs.info('[DRY RUN] Would delete: ' + gr.number + ' (' + rowCount + ' rows)');
      } else {
        // Delete rows first
        var rowGr = new GlideRecord('sys_import_set_row');
        rowGr.addQuery('sys_import_set', gr.sys_id);
        rowGr.deleteMultiple();

        // Delete import set
        gr.deleteRecord();
        stats.rows += rowCount;
      }

      stats.sets++;
    }

    gs.info('Cleanup complete: ' + stats.sets + ' import sets, ' + stats.rows + ' rows');
  description: Cleanup old import sets

步骤 9.3：归档导入历史记录

清理前保留导入历史记录。

Tool: SN-Execute-Background-Script
Parameters:
  script: |
    // Archive import history before cleanup
    var importSetId = '[import_set_sys_id]';

    var importSet = new GlideRecord('sys_import_set');
    if (importSet.get(importSetId)) {

      // Get statistics
      var stats = {
        number: importSet.number.toString(),
        table_name: importSet.table_name.toString(),
        state: importSet.state.toString(),
        sys_created_on: importSet.sys_created_on.toString(),
        sys_created_by: importSet.sys_created_by.toString()
      };

      // Count by state
      var ga = new GlideAggregate('sys_import_set_row');
      ga.addQuery('sys_import_set', importSetId);
      ga.addAggregate('COUNT');
      ga.groupBy('sys_import_state');
      ga.query();

      stats.results = {};
      while (ga.next()) {
        stats.results[ga.sys_import_state.toString()] = parseInt(ga.getAggregate('COUNT'));
      }

      // Create archive record
      var archive = new GlideRecord('u_import_archive');
      archive.initialize();
      archive.u_import_set_number = stats.number;
      archive.u_source_table = stats.table_name;
      archive.u_import_date = stats.sys_created_on;
      archive.u_imported_by = stats.sys_created_by;
      archive.u_statistics = JSON.stringify(stats);
      archive.insert();

      gs.info('Archived import set: ' + stats.number);
    }
  description: Archive import set history

工具使用摘要

操作	MCP 工具	用途
创建数据源	SN-Create-Record	配置导入源
创建转换映射	SN-Create-Record	定义字段映射
创建字段映射	SN-Batch-Create	将源字段映射到目标字段
创建脚本	SN-Create-Record	添加转换逻辑
查询导入行	SN-Query-Table	监控导入进度
重试错误	SN-Execute-Background-Script	重新处理失败的行
性能检查	SN-Execute-Background-Script	监控吞吐量
回滚	SN-Execute-Background-Script	撤销导入

性能指南

数据量	推荐方法	预计耗时
< 1,000	标准导入	< 1 分钟
1,000 - 10,000	标准导入	1-10 分钟
10,000 - 100,000	分块导入	10-60 分钟
100,000+	后台作业 + 分块处理	数小时

性能提示：

• 为合并字段创建索引
• 批量导入时禁用业务规则
• 对大型数据集使用分块处理
• 在非高峰时段安排导入
• 导入期间监控系统性能

最佳实践

• 始终先测试： 在预生产环境中使用示例数据运行导入
• 使用合并字段： 通过正确匹配来防止记录重复
• 验证数据： 使用 onBefore 脚本在转换前验证数据
• 妥善处理错误： 不要遇到第一个错误就中止，应记录错误并继续
• 监控进度： 跟踪行数和处理时间
• 清理数据： 删除旧的导入集以保持性能
• 记录映射： 保存字段映射和转换的记录
• 版本控制： 将转换脚本存储在源代码管理中
• 审计跟踪： 记录所有导入操作以满足合规要求

故障排除

转换因“No Coalesce”而失败

症状： 所有行都标记为错误，并显示“Multiple records found” 原因： 合并字段匹配到多条记录 解决方案：

• 添加更多合并字段，以实现唯一匹配
• 设置 multi_coalesce_action 来处理重复记录
• 导入前清理目标表中的重复记录

导入超时

症状： 导入在处理中途停止 原因： 单次导入的记录过多 解决方案：

• 使用分块处理
• 增加事务超时（系统属性）
• 暂时禁用业务规则

引用字段未映射

症状： 转换后引用字段为空 原因： 未找到被引用的记录 解决方案：

• 使用 onForeignInsert 脚本创建缺失的引用记录
• 确认引用字段查询与源数据匹配
• 检查引用表中是否存在匹配的记录

性能下降

症状： 导入速度随时间显著降低 原因： 表碎片化、缺少索引 解决方案：

• 为合并字段添加索引
• 执行表维护
• 归档旧的导入集

数据截断

症状： 字段值被截断 原因： 源数据超出字段长度 解决方案：

• 检查架构中的字段 max_length
• 在 onBefore 脚本中截断或转换数据
• 如适用，增加字段长度

示例

示例 1：基本用户导入

# 1. Create transform map
Tool: SN-Create-Record
Parameters:
  table_name: sys_transform_map
  data:
    name: Basic User Import
    source_table: u_user_import
    target_table: sys_user
    active: true

# 2. Create field mappings
Tool: SN-Batch-Create
Parameters:
  records:
    - table_name: sys_transform_entry
      data:
        map: [transform_map_sys_id]
        source_field: u_email
        target_field: email
        coalesce: true
    - table_name: sys_transform_entry
      data:
        map: [transform_map_sys_id]
        source_field: u_first_name
        target_field: first_name
    - table_name: sys_transform_entry
      data:
        map: [transform_map_sys_id]
        source_field: u_last_name
        target_field: last_name

示例 2：带关系的 CI 导入

# Transform map for CI import with relationship creation
Tool: SN-Create-Record
Parameters:
  table_name: sys_transform_script
  data:
    map: [ci_transform_map_sys_id]
    script_type: onAfter
    script: |
      // Create CI relationships after import
      if (action == 'insert' && source.u_parent_ci) {
        var rel = new GlideRecord('cmdb_rel_ci');
        rel.initialize();
        rel.parent = lookupCI(source.u_parent_ci);
        rel.child = target.sys_id;
        rel.type = getRelationType('Depends on::Used by');
        rel.insert();
      }

      function lookupCI(name) {
        var ci = new GlideRecord('cmdb_ci');
        ci.addQuery('name', name);
        ci.query();
        return ci.next() ? ci.sys_id : '';
      }

      function getRelationType(name) {
        var type = new GlideRecord('cmdb_rel_type');
        type.addQuery('name', name);
        type.query();
        return type.next() ? type.sys_id : '';
      }

示例 3：增量数据同步

# Scheduled import for incremental sync
Tool: SN-Create-Record
Parameters:
  table_name: sysauto_script
  data:
    name: Incremental User Sync
    script: |
      // Get last sync timestamp
      var prop = gs.getProperty('user.import.last_sync', '1970-01-01 00:00:00');
      var lastSync = new GlideDateTime();
      lastSync.setDisplayValue(prop);

      // Update data source query with timestamp
      var ds = new GlideRecord('sys_data_source');
      ds.get('[data_source_sys_id]');
      ds.query = "SELECT * FROM employees WHERE modified_date > '" + lastSync.getValue() + "'";
      ds.update();

      // Run import
      var loader = new GlideImportSetLoader();
      var importSetId = loader.loadAttachment('[data_source_sys_id]');

      var transformer = new GlideImportSetTransformer();
      transformer.setImportSetID(importSetId);
      transformer.setTransformMapID('[transform_map_sys_id]');
      transformer.transform();

      // Update last sync timestamp
      gs.setProperty('user.import.last_sync', new GlideDateTime().getDisplayValue());
    run_type: periodically
    run_period: 60
    active: true

相关技能

• admin/batch-operations - 批量数据操作
• admin/generic-crud-operations - 基本表操作
• admin/script-execution - 后台脚本执行
• admin/schema-discovery - 表结构发现
• cmdb/data-quality - 数据质量验证

参考资料

• ServiceNow 导入集
• 转换映射
• 数据源
• 转换脚本
• 导入集最佳实践