创建作业规范文件

配置

关系目标配置选项及其默认值

{
  ...
  "active": true,
  "source_transformations": {
    "enable_grouping": true
  },
  "depends_on": ["<dependencyTargetName1>", "<dependencyTargetName2>", ...]
}

• active (bool) — 是否应将该目标包含在导入中。

• source_transformations (object) — 如果 enable_grouping 设置为 true，导入程序将对 key_constraints 和 properties 中指定的所有字段执行 SQL GROUP BY。如果设置为 false，则源中的任何重复数据都将被推送到 Neo4j，这可能会导致约束错误或降低插入效率。该对象还可以包含聚合函数和其他字段，请参阅 源转换。

• depends_on (字符串列表) — 应在当前目标 之前 执行的目标 name。

示例

用于导入 ACTED_IN 关系的关系对象示例

{
  "source": "acted_in",
  "name": "Acted_in",
  "type": "ACTED_IN",
  "write_mode": "merge",
  "node_match_mode": "match",
  "start_node_reference": "Persons",
  "end_node_reference": "Movies",
  "properties": [
    {
      "source_field": "role",
      "target_field": "role",
      "target_property_type": "string"
    }
  ]
}

---

自定义查询目标

当导入需要无法轻松融入节点/关系目标格式的复杂查询时，自定义查询目标非常有用。查询目标通过变量 $rows 接收行批次。

自定义查询必须在 targets 对象内的 queries 键下进行分组。

查询目标规范框架

"targets": {
  ...
  "queries": [
    { <querySpec1> },
    { <querySpec2> },
    ...
  ]
}

不要使用自定义查询来运行不直接依赖于源的 Cypher；请改用 actions。一次性查询（特别是如果不是幂等的）不适合在自定义查询目标中使用。原因在于目标中的查询是分批运行的，因此自定义查询可能会根据从源中提取的 $rows 批次数量多次运行。

必填字段

每个查询目标至少必须包含属性 source, name 和 query。

{
  "source": "<sourceName>",
  "name": "<targetName>",
  "query": "<cypherQuery>"
}

• source (string) — 此目标应从中获取数据的源名称。应与 sources 对象中的名称之一匹配。

• name (string) — 目标的人性化名称（在所有名称中唯一）。

• query (string) — 一个 Cypher 查询。来自源的数据可作为参数 $rows 中的列表使用。

配置

查询目标配置选项及其默认值

{
  ...
  "active": true,
  "depends_on": ["<dependencyTargetName1>", "<dependencyTargetName2>", ...]
}

• active (bool) — 是否应将该目标包含在导入中。

• depends_on (字符串列表) — 应在当前目标 之前 执行的目标 name。

示例

用于导入 Person 节点并设置创建日期的查询对象示例

{
  "custom_query": {
    "name": "Person nodes",
    "source": "persons",
    "query": "UNWIND $rows AS row WHERE row.person_tmdbId IS NOT NULL MERGE (p:Person {id: row.person_tmdbId, name: row.name, born_in: row.bornIn, born: date(row.born), died: date(row.died)}) ON CREATE SET p.created_time=datetime()"
  }
}

源转换

每个节点和关系目标都可以选择性地包含一个包含聚合函数的 source_transformation 属性。这对于从更细粒度的源提取更高级别的维度非常有用。聚合会产生可用于属性映射的额外字段。

"source_transformations": {
  "enable_grouping": true,
  "aggregations": [ {
    "expression": "",
    "field_name": ""
   },
   { aggregationObj2 }, ...
  ],
  "limit": -1,
  "where": "",
  "order_by": [
    {
      "expression": "column_name",
      "order": "<asc/desc>"
    },
    { orderObj2 }, ...
  ],
}

• enable_grouping (bool) — 必须为 true 才能使 aggregations/where 工作。

• aggregations (对象列表) — 聚合在 expression 属性中指定为 SQL 查询，结果以 field_name 中指定的名称作为源列提供。

• limit (int) — 限制考虑用于导入的源行数（默认为无限制，编码为 -1）。

• where (string) — 在导入前过滤掉源数据（使用 SQL WHERE 子句格式）。

• order_by (对象列表) — 对源强制执行排序。

示例

虚拟数据集上的转换对象示例

{
  "enable_grouping": true,
  "aggregations": [
    {
      "expression": "SUM(unit_price*quantity)",
      "field_name": "total_amount_sold"
    },
    {
      "expression": "SUM(quantity)",
      "field_name": "total_quantity_sold"
    }
  ],
  "limit": 50,
  "where": "sourceId IS NOT NULL"
}

默认情况下，源只被处理一次；其数据随后被分发到各个目标。但是，带有源转换的目标会触发新的数据获取，因为生成的源查询与默认查询不同。因此，对于没有转换的目标，源会被处理一次，对于定义了转换的目标，则会额外处理多次。因此，原始源查询必须是确定性的，否则不同的目标可能会接收到不同的数据。

操作 (Actions)

actions 部分包含可以在导入过程的特定步骤之前或之后运行的命令。每个步骤称为一个 stage。例如，您可以在步骤完成时提交 HTTP 请求、在源上执行 SQL 查询或在 Neo4j 目标实例上运行 Cypher 语句。

操作规范框架

  ...
  "actions": [
    { <actionSpec1> },
    { <actionSpec2> },
    ...
  ]

每个操作对象至少必须包含属性 name, type 和 stage。其他属性取决于操作类型。

{
  "type": "http",
  "name": "<actionName>",
  "stage": "<stageName>",
  "method": "<get/post>",
  "url": "<targetUrl>",
  "headers": {}
}

• type (string) — 操作类型。

• name (string) — 操作的人性化名称（在所有名称中唯一）。

• stage (string) — 操作应在导入的哪个时间点运行。有效值为：start, post_sources, pre_nodes, post_nodes, pre_relationships, post_relationships, pre_queries, post_queries, end。

• method (string) — HTTP 方法；可以是 get 或 post。

• url (string) — HTTP 请求应指向的 URL。

• headers (object, 可选) — 请求头。

导入完成后发送 GET 请求的操作示例

{
  "type": "http",
  "name": "Post load ping",
  "stage": "end",
  "method": "get",
  "url": "/success",
  "headers": {
    "secret": "314159",
    "moreSecret": "17320"
  }
}

{
  "type": "cypher",
  "name": "<actionName>",
  "stage": "<stageName>",
  "query": "<cypherQuery>",
  "execution_mode": "<transaction/autocommit>"
}

• type (string) — 操作类型。

• name (string) — 操作的人性化名称（在所有名称中唯一）。

• stage (string) — 操作应在导入的哪个时间点运行。有效值为：start, post_sources, pre_nodes, post_nodes, pre_relationships, post_relationships, pre_queries, post_queries, end。

• query (string) — 要运行的 Cypher 查询。

• execution_mode (string, 可选) — 查询应在何种模式下执行。有效值为 transaction, autocommit (默认: transaction)。

导入完成后创建 importJob 节点的操作示例

{
  "type": "cypher",
  "name": "Post load log",
  "stage": "end",
  "query": "MERGE (:importJob {date: datetime()})"
}

{
  "type": "bigquery",
  "name": "<actionName>",
  "stage": "<stageName>",
  "sql": "<sqlQuery>"
}

• type (string) — 操作类型。

• name (string) — 操作的人性化名称（在所有名称中唯一）。

• stage (string) — 操作应在导入的哪个时间点运行。有效值为：start, post_sources, pre_nodes, post_nodes, pre_relationships, post_relationships, pre_queries, post_queries, end。

• sql (string) — 要运行的 SQL 查询。

导入完成后发送 GET 请求的操作示例

{
  "type": "bigquery",
  "name": "Post load log",
  "stage": "end",
  "sql": "INSERT INTO logs.imports (time) VALUES (NOW())"
}

变量

尚不支持变量。