Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 dependabot[bot]
|
fbb239e44e |
@@ -29,6 +29,12 @@
|
||||
<artifactId>datax-common</artifactId>
|
||||
<version>${datax-project-version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.alibaba.datax</groupId>
|
||||
<artifactId>simulator</artifactId>
|
||||
<version>${datax-project-version}</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.slf4j</groupId>
|
||||
<artifactId>slf4j-api</artifactId>
|
||||
|
||||
+1
@@ -12,6 +12,7 @@ import com.alibaba.datax.plugin.rdbms.util.DataBaseType;
|
||||
import com.alibaba.datax.plugin.rdbms.writer.CommonRdbmsWriter;
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import com.alibaba.fastjson.JSONArray;
|
||||
import ru.yandex.clickhouse.ClickHouseTuple;
|
||||
|
||||
import java.sql.Array;
|
||||
import java.sql.Connection;
|
||||
|
||||
Regular → Executable
+1
-1
@@ -36,7 +36,7 @@
|
||||
<dependency>
|
||||
<groupId>commons-beanutils</groupId>
|
||||
<artifactId>commons-beanutils</artifactId>
|
||||
<version>1.9.4</version>
|
||||
<version>1.9.2</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.httpcomponents</groupId>
|
||||
|
||||
+2
-2
@@ -435,7 +435,7 @@ DataX的内部类型在实现上会选用不同的java类型：
|
||||
|
||||
#### 如何处理脏数据
|
||||
|
||||
在`Reader.Task`和`Writer.Task`中，通过`AbstractTaskPlugin.getTaskPluginCollector()`可以拿到一个`TaskPluginCollector`，它提供了一系列`collectDirtyRecord`的方法。当脏数据出现时，只需要调用合适的`collectDirtyRecord`方法，把被认为是脏数据的`Record`传入即可。
|
||||
在`Reader.Task`和`Writer.Task`中，功过`AbstractTaskPlugin.getPluginCollector()`可以拿到一个`TaskPluginCollector`，它提供了一系列`collectDirtyRecord`的方法。当脏数据出现时，只需要调用合适的`collectDirtyRecord`方法，把被认为是脏数据的`Record`传入即可。
|
||||
|
||||
用户可以在任务的配置中指定脏数据限制条数或者百分比限制，当脏数据超出限制时，框架会结束同步任务，退出。插件需要保证脏数据都被收集到，其他工作交给框架就好。
|
||||
|
||||
@@ -468,4 +468,4 @@ DataX的内部类型在实现上会选用不同的java类型：
|
||||
- 测试参数集（多组），系统参数（比如并发数），插件参数（比如batchSize）
|
||||
- 不同参数下同步速度（Rec/s, MB/s），机器负载（load, cpu）等，对数据源压力（load, cpu, mem等）。
|
||||
6. **约束限制**：是否存在其他的使用限制条件。
|
||||
7. **FAQ**：用户经常会遇到的问题。
|
||||
7. **FQA**：用户经常会遇到的问题。
|
||||
|
||||
@@ -63,7 +63,6 @@ FtpWriter实现了从DataX协议转为FTP文件功能，FTP文件本身是无结
|
||||
"nullFormat": "null",
|
||||
"dateFormat": "yyyy-MM-dd",
|
||||
"fileFormat": "csv",
|
||||
"suffix": ".csv",
|
||||
"header": []
|
||||
}
|
||||
}
|
||||
@@ -201,14 +200,6 @@ FtpWriter实现了从DataX协议转为FTP文件功能，FTP文件本身是无结
|
||||
* 必选：否 <br />
|
||||
|
||||
* 默认值：text <br />
|
||||
|
||||
* **suffix**
|
||||
|
||||
* 描述：最后输出文件的后缀，当前支持 ".text"以及".csv"
|
||||
|
||||
* 必选：否 <br />
|
||||
|
||||
* 默认值："" <br />
|
||||
|
||||
* **header**
|
||||
|
||||
|
||||
@@ -1,260 +0,0 @@
|
||||
|
||||
# DataX GDBReader
|
||||
|
||||
## 1. 快速介绍
|
||||
|
||||
GDBReader插件实现读取GDB实例数据的功能，通过`Gremlin Client`连接远程GDB实例，按配置提供的`label`生成查询DSL，遍历点或边数据，包括属性数据，并将数据写入到Record中给到Writer使用。
|
||||
|
||||
## 2. 实现原理
|
||||
|
||||
GDBReader使用`Gremlin Client`连接GDB实例，按`label`分不同Task取点或边数据。
|
||||
单个Task中按`label`遍历点或边的id，再切分范围分多次请求查询点或边和属性数据，最后将点或边数据根据配置转换成指定格式记录发送给下游写插件。
|
||||
|
||||
GDBReader按`label`切分多个Task并发，同一个`label`的数据批量异步获取来加快读取速度。如果配置读取的`label`列表为空，任务启动前会从GDB查询所有`label`再切分Task。
|
||||
|
||||
## 3. 功能说明
|
||||
|
||||
GDB中点和边不同，读取需要区分点和边点配置。
|
||||
|
||||
### 3.1 点配置样例
|
||||
|
||||
```
|
||||
{
|
||||
"job": {
|
||||
"setting": {
|
||||
"speed": {
|
||||
"channel": 1
|
||||
}
|
||||
"errorLimit": {
|
||||
"record": 1
|
||||
}
|
||||
},
|
||||
|
||||
"content": [
|
||||
{
|
||||
"reader": {
|
||||
"name": "gdbreader",
|
||||
"parameter": {
|
||||
"host": "10.218.145.24",
|
||||
"port": 8182,
|
||||
"username": "***",
|
||||
"password": "***",
|
||||
"fetchBatchSize": 100,
|
||||
"rangeSplitSize": 1000,
|
||||
"labelType": "VERTEX",
|
||||
"labels": ["label1", "label2"],
|
||||
"column": [
|
||||
{
|
||||
"name": "id",
|
||||
"type": "string",
|
||||
"columnType": "primaryKey"
|
||||
},
|
||||
{
|
||||
"name": "label",
|
||||
"type": "string",
|
||||
"columnType": "primaryLabel"
|
||||
},
|
||||
{
|
||||
"name": "age",
|
||||
"type": "int",
|
||||
"columnType": "vertexProperty"
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
"writer": {
|
||||
"name": "streamwriter",
|
||||
"parameter": {
|
||||
"print": true
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 3.2 边配置样例
|
||||
|
||||
```
|
||||
{
|
||||
"job": {
|
||||
"setting": {
|
||||
"speed": {
|
||||
"channel": 1
|
||||
},
|
||||
"errorLimit": {
|
||||
"record": 1
|
||||
}
|
||||
},
|
||||
|
||||
"content": [
|
||||
{
|
||||
"reader": {
|
||||
"name": "gdbreader",
|
||||
"parameter": {
|
||||
"host": "10.218.145.24",
|
||||
"port": 8182,
|
||||
"username": "***",
|
||||
"password": "***",
|
||||
"fetchBatchSize": 100,
|
||||
"rangeSplitSize": 1000,
|
||||
"labelType": "EDGE",
|
||||
"labels": ["label1", "label2"],
|
||||
"column": [
|
||||
{
|
||||
"name": "id",
|
||||
"type": "string",
|
||||
"columnType": "primaryKey"
|
||||
},
|
||||
{
|
||||
"name": "label",
|
||||
"type": "string",
|
||||
"columnType": "primaryLabel"
|
||||
},
|
||||
{
|
||||
"name": "srcId",
|
||||
"type": "string",
|
||||
"columnType": "srcPrimaryKey"
|
||||
},
|
||||
{
|
||||
"name": "srcLabel",
|
||||
"type": "string",
|
||||
"columnType": "srcPrimaryLabel"
|
||||
},
|
||||
{
|
||||
"name": "dstId",
|
||||
"type": "string",
|
||||
"columnType": "srcPrimaryKey"
|
||||
},
|
||||
{
|
||||
"name": "dstLabel",
|
||||
"type": "string",
|
||||
"columnType": "srcPrimaryLabel"
|
||||
},
|
||||
{
|
||||
"name": "name",
|
||||
"type": "string",
|
||||
"columnType": "edgeProperty"
|
||||
},
|
||||
{
|
||||
"name": "weight",
|
||||
"type": "double",
|
||||
"columnType": "edgeProperty"
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
|
||||
"writer": {
|
||||
"name": "streamwriter",
|
||||
"parameter": {
|
||||
"print": true
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 3.3 参数说明
|
||||
|
||||
* **host**
|
||||
* 描述：GDB实例连接地址，对应'实例管理'->'基本信息'页面的网络地址
|
||||
* 必选：是
|
||||
* 默认值：无
|
||||
|
||||
* **port**
|
||||
* 描述：GDB实例连接地址对应的端口
|
||||
* 必选：是
|
||||
* 默认值：8182
|
||||
|
||||
* **username**
|
||||
* 描述：GDB实例账号名
|
||||
* 必选：是
|
||||
* 默认值：无
|
||||
|
||||
* **password**
|
||||
* 描述：GDB实例账号名对应的密码
|
||||
* 必选：是
|
||||
* 默认值：无
|
||||
|
||||
* **fetchBatchSize**
|
||||
* 描述：一次GDB请求读取点或边的数量，响应包含点或边以及属性
|
||||
* 必选：是
|
||||
* 默认值：100
|
||||
|
||||
* **rangeSplitSize**
|
||||
* 描述：id遍历，一次遍历请求扫描的id个数
|
||||
* 必选：是
|
||||
* 默认值：10 \* fetchBatchSize
|
||||
|
||||
* **labels**
|
||||
* 描述：标签数组，即需要导出的点或边标签，支持读取多个标签，用数组表示。如果留空([])，表示GDB中所有点或边标签
|
||||
* 必选：是
|
||||
* 默认值：无
|
||||
|
||||
* **labelType**
|
||||
* 描述：数据标签类型，支持点、边两种枚举值
|
||||
* VERTEX：表示点
|
||||
* EDGE：表示边
|
||||
* 必选：是
|
||||
* 默认值：无
|
||||
|
||||
* **column**
|
||||
* 描述：点或边字段映射关系配置
|
||||
* 必选：是
|
||||
* 默认值：无
|
||||
|
||||
* **column -> name**
|
||||
* 描述：点或边映射关系的字段名，指定属性时表示读取的属性名，读取其他字段时会被忽略
|
||||
* 必选：是
|
||||
* 默认值：无
|
||||
|
||||
* **column -> type**
|
||||
* 描述：点或边映射关系的字段类型
|
||||
* id, label在GDB中都是string类型，配置非string类型时可能会转换失败
|
||||
* 普通属性支持基础类型，包括int, long, float, double, boolean, string
|
||||
* GDBReader尽量将读取到的数据转换成配置要求的类型，但转换失败会导致该条记录错误
|
||||
* 必选：是
|
||||
* 默认值：无
|
||||
|
||||
* **column -> columnType**
|
||||
* 描述：GDB点或边数据到列数据的映射关系，支持以下枚举值：
|
||||
* primaryKey： 表示该字段是点或边的id
|
||||
* primaryLabel： 表示该字段是点或边的label
|
||||
* srcPrimaryKey： 表示该字段是边关联的起点id，只在读取边时使用
|
||||
* srcPrimaryLabel： 表示该字段是边关联的起点label，只在读取边时使用
|
||||
* dstPrimaryKey： 表示该字段是边关联的终点id，只在读取边时使用
|
||||
* dstPrimaryLabel： 表示该字段是边关联的终点label，只在读取边时使用
|
||||
* vertexProperty： 表示该字段是点的属性，只在读取点时使用，应用到SET属性时只读取其中的一个属性值
|
||||
* vertexJsonProperty： 表示该字段是点的属性集合，只在读取点时使用。属性集合使用JSON格式输出，包含所有的属性，不能与其他vertexProperty配置一起使用
|
||||
* edgeProperty： 表示该字段是边的属性，只在读取边时使用
|
||||
* edgeJsonProperty： 表示该字段是边的属性集合，只在读取边时使用。属性集合使用JSON格式输出，包含所有的属性，不能与其他edgeProperty配置一起使用
|
||||
* 必选：是
|
||||
* 默认值：无
|
||||
* vertexJsonProperty格式示例，新增`c`字段区分SET属性，但是SET属性只包含单个属性值时会标记成普通属性
|
||||
```
|
||||
{"properties":[
|
||||
{"k":"name","t","string","v":"Jack","c":"set"},
|
||||
{"k":"name","t","string","v":"Luck","c":"set"},
|
||||
{"k":"age","t","int","v":"20","c":"single"}
|
||||
]}
|
||||
```
|
||||
* edgeJsonProperty格式示例，边不支持多值属性
|
||||
```
|
||||
{"properties":[
|
||||
{"k":"created_at","t","long","v":"153498653"},
|
||||
{"k":"weight","t","double","v":"3.14"}
|
||||
]}
|
||||
|
||||
## 4 性能报告
|
||||
(TODO)
|
||||
|
||||
## 5 使用约束
|
||||
无
|
||||
|
||||
## 6 FAQ
|
||||
无
|
||||
|
||||
@@ -1,125 +0,0 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<parent>
|
||||
<artifactId>datax-all</artifactId>
|
||||
<groupId>com.alibaba.datax</groupId>
|
||||
<version>0.0.1-SNAPSHOT</version>
|
||||
</parent>
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<artifactId>gdbreader</artifactId>
|
||||
|
||||
<groupId>com.alibaba.datax</groupId>
|
||||
<version>0.0.1-SNAPSHOT</version>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>com.alibaba.datax</groupId>
|
||||
<artifactId>datax-common</artifactId>
|
||||
<version>${datax-project-version}</version>
|
||||
<exclusions>
|
||||
<exclusion>
|
||||
<artifactId>slf4j-log4j12</artifactId>
|
||||
<groupId>org.slf4j</groupId>
|
||||
</exclusion>
|
||||
</exclusions>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.alibaba.datax</groupId>
|
||||
<artifactId>datax-core</artifactId>
|
||||
<version>${datax-project-version}</version>
|
||||
<scope>test</scope>
|
||||
<exclusions>
|
||||
<exclusion>
|
||||
<artifactId>slf4j-log4j12</artifactId>
|
||||
<groupId>org.slf4j</groupId>
|
||||
</exclusion>
|
||||
</exclusions>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.slf4j</groupId>
|
||||
<artifactId>slf4j-api</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>ch.qos.logback</groupId>
|
||||
<artifactId>logback-classic</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.apache.tinkerpop</groupId>
|
||||
<artifactId>gremlin-driver</artifactId>
|
||||
<version>3.4.1</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.projectlombok</groupId>
|
||||
<artifactId>lombok</artifactId>
|
||||
<version>1.18.8</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.junit.jupiter</groupId>
|
||||
<artifactId>junit-jupiter-api</artifactId>
|
||||
<version>5.4.0</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.junit.jupiter</groupId>
|
||||
<artifactId>junit-jupiter-engine</artifactId>
|
||||
<version>5.4.0</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<!-- compiler plugin -->
|
||||
<plugin>
|
||||
<artifactId>maven-compiler-plugin</artifactId>
|
||||
<configuration>
|
||||
<source>1.6</source>
|
||||
<target>1.6</target>
|
||||
<encoding>${project-sourceEncoding}</encoding>
|
||||
</configuration>
|
||||
</plugin>
|
||||
<!-- test case plugin -->
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-surefire-plugin</artifactId>
|
||||
<version>2.22.0</version>
|
||||
<configuration>
|
||||
<includes>
|
||||
<include>**/*Test*.class</include>
|
||||
</includes>
|
||||
</configuration>
|
||||
</plugin>
|
||||
<!-- assembly plugin -->
|
||||
<plugin>
|
||||
<artifactId>maven-assembly-plugin</artifactId>
|
||||
<configuration>
|
||||
<descriptors>
|
||||
<descriptor>src/main/assembly/package.xml</descriptor>
|
||||
</descriptors>
|
||||
<finalName>datax</finalName>
|
||||
</configuration>
|
||||
<executions>
|
||||
<execution>
|
||||
<id>dwzip</id>
|
||||
<phase>package</phase>
|
||||
<goals>
|
||||
<goal>single</goal>
|
||||
</goals>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-compiler-plugin</artifactId>
|
||||
<configuration>
|
||||
<source>8</source>
|
||||
<target>8</target>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</project>
|
||||
@@ -1,35 +0,0 @@
|
||||
<assembly
|
||||
xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
|
||||
<id></id>
|
||||
<formats>
|
||||
<format>dir</format>
|
||||
</formats>
|
||||
<includeBaseDirectory>false</includeBaseDirectory>
|
||||
<fileSets>
|
||||
<fileSet>
|
||||
<directory>src/main/resources</directory>
|
||||
<includes>
|
||||
<include>plugin.json</include>
|
||||
<include>plugin_job_template.json</include>
|
||||
</includes>
|
||||
<outputDirectory>plugin/reader/gdbreader</outputDirectory>
|
||||
</fileSet>
|
||||
<fileSet>
|
||||
<directory>target/</directory>
|
||||
<includes>
|
||||
<include>gdbreader-0.0.1-SNAPSHOT.jar</include>
|
||||
</includes>
|
||||
<outputDirectory>plugin/reader/gdbreader</outputDirectory>
|
||||
</fileSet>
|
||||
</fileSets>
|
||||
|
||||
<dependencySets>
|
||||
<dependencySet>
|
||||
<useProjectArtifact>false</useProjectArtifact>
|
||||
<outputDirectory>plugin/reader/gdbreader/libs</outputDirectory>
|
||||
<scope>runtime</scope>
|
||||
</dependencySet>
|
||||
</dependencySets>
|
||||
</assembly>
|
||||
@@ -1,231 +0,0 @@
|
||||
package com.alibaba.datax.plugin.reader.gdbreader;
|
||||
|
||||
import com.alibaba.datax.common.element.Record;
|
||||
import com.alibaba.datax.common.exception.DataXException;
|
||||
import com.alibaba.datax.common.plugin.RecordSender;
|
||||
import com.alibaba.datax.common.spi.Reader;
|
||||
import com.alibaba.datax.common.util.Configuration;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.mapping.DefaultGdbMapper;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.mapping.MappingRule;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.mapping.MappingRuleFactory;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.model.GdbElement;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.model.GdbGraph;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.model.ScriptGdbGraph;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.util.ConfigHelper;
|
||||
import org.apache.tinkerpop.gremlin.driver.ResultSet;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
|
||||
public class GdbReader extends Reader {
|
||||
private final static int DEFAULT_FETCH_BATCH_SIZE = 200;
|
||||
private static GdbGraph graph;
|
||||
private static Key.ExportType exportType;
|
||||
|
||||
/**
|
||||
* Job 中的方法仅执行一次，Task 中方法会由框架启动多个 Task 线程并行执行。
|
||||
* <p/>
|
||||
* 整个 Reader 执行流程是：
|
||||
* <pre>
|
||||
* Job类init-->prepare-->split
|
||||
*
|
||||
* Task类init-->prepare-->startRead-->post-->destroy
|
||||
* Task类init-->prepare-->startRead-->post-->destroy
|
||||
*
|
||||
* Job类post-->destroy
|
||||
* </pre>
|
||||
*/
|
||||
public static class Job extends Reader.Job {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(Job.class);
|
||||
|
||||
private Configuration jobConfig = null;
|
||||
|
||||
@Override
|
||||
public void init() {
|
||||
this.jobConfig = super.getPluginJobConf();
|
||||
|
||||
/**
|
||||
* 注意：此方法仅执行一次。
|
||||
* 最佳实践：通常在这里对用户的配置进行校验：是否缺失必填项？有无错误值？有没有无关配置项？...
|
||||
* 并给出清晰的报错/警告提示。校验通常建议采用静态工具类进行，以保证本类结构清晰。
|
||||
*/
|
||||
|
||||
ConfigHelper.assertGdbClient(jobConfig);
|
||||
ConfigHelper.assertLabels(jobConfig);
|
||||
try {
|
||||
exportType = Key.ExportType.valueOf(jobConfig.getString(Key.EXPORT_TYPE));
|
||||
} catch (NullPointerException | IllegalArgumentException e) {
|
||||
throw DataXException.asDataXException(GdbReaderErrorCode.BAD_CONFIG_VALUE, Key.EXPORT_TYPE);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void prepare() {
|
||||
/**
|
||||
* 注意：此方法仅执行一次。
|
||||
* 最佳实践：如果 Job 中有需要进行数据同步之前的处理，可以在此处完成，如果没有必要则可以直接去掉。
|
||||
*/
|
||||
|
||||
try {
|
||||
graph = new ScriptGdbGraph(jobConfig, exportType);
|
||||
} catch (RuntimeException e) {
|
||||
throw DataXException.asDataXException(GdbReaderErrorCode.FAIL_CLIENT_CONNECT, e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<Configuration> split(int adviceNumber) {
|
||||
/**
|
||||
* 注意：此方法仅执行一次。
|
||||
* 最佳实践：通常采用工具静态类完成把 Job 配置切分成多个 Task 配置的工作。
|
||||
* 这里的 adviceNumber 是框架根据用户的同步速度的要求建议的切分份数，仅供参考，不是强制必须切分的份数。
|
||||
*/
|
||||
List<String> labels = ConfigHelper.assertLabels(jobConfig);
|
||||
|
||||
/**
|
||||
* 配置label列表为空时，尝试查询GDB中所有label，添加到读取列表
|
||||
*/
|
||||
if (labels.isEmpty()) {
|
||||
try {
|
||||
labels.addAll(graph.getLabels().keySet());
|
||||
} catch (RuntimeException ex) {
|
||||
throw DataXException.asDataXException(GdbReaderErrorCode.FAIL_FETCH_LABELS, ex.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
if (labels.isEmpty()) {
|
||||
throw DataXException.asDataXException(GdbReaderErrorCode.FAIL_FETCH_LABELS, "none labels to read");
|
||||
}
|
||||
|
||||
return ConfigHelper.splitConfig(jobConfig, labels);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void post() {
|
||||
/**
|
||||
* 注意：此方法仅执行一次。
|
||||
* 最佳实践：如果 Job 中有需要进行数据同步之后的后续处理，可以在此处完成。
|
||||
*/
|
||||
}
|
||||
|
||||
@Override
|
||||
public void destroy() {
|
||||
/**
|
||||
* 注意：此方法仅执行一次。
|
||||
* 最佳实践：通常配合 Job 中的 post() 方法一起完成 Job 的资源释放。
|
||||
*/
|
||||
try {
|
||||
graph.close();
|
||||
} catch (Exception ex) {
|
||||
LOG.error("Failed to close client : {}", ex);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class Task extends Reader.Task {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(Task.class);
|
||||
private static MappingRule rule;
|
||||
private Configuration taskConfig;
|
||||
private String fetchLabel = null;
|
||||
|
||||
private int rangeSplitSize;
|
||||
private int fetchBatchSize;
|
||||
|
||||
@Override
|
||||
public void init() {
|
||||
this.taskConfig = super.getPluginJobConf();
|
||||
|
||||
/**
|
||||
* 注意：此方法每个 Task 都会执行一次。
|
||||
* 最佳实践：此处通过对 taskConfig 配置的读取，进而初始化一些资源为 startRead()做准备。
|
||||
*/
|
||||
fetchLabel = taskConfig.getString(Key.LABEL);
|
||||
fetchBatchSize = taskConfig.getInt(Key.FETCH_BATCH_SIZE, DEFAULT_FETCH_BATCH_SIZE);
|
||||
rangeSplitSize = taskConfig.getInt(Key.RANGE_SPLIT_SIZE, fetchBatchSize * 10);
|
||||
rule = MappingRuleFactory.getInstance().create(taskConfig, exportType);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void prepare() {
|
||||
/**
|
||||
* 注意：此方法仅执行一次。
|
||||
* 最佳实践：如果 Job 中有需要进行数据同步之后的处理，可以在此处完成，如果没有必要则可以直接去掉。
|
||||
*/
|
||||
}
|
||||
|
||||
@Override
|
||||
public void startRead(RecordSender recordSender) {
|
||||
/**
|
||||
* 注意：此方法每个 Task 都会执行一次。
|
||||
* 最佳实践：此处适当封装确保简洁清晰完成数据读取工作。
|
||||
*/
|
||||
|
||||
String start = "";
|
||||
while (true) {
|
||||
List<String> ids;
|
||||
try {
|
||||
ids = graph.fetchIds(fetchLabel, start, rangeSplitSize);
|
||||
if (ids.isEmpty()) {
|
||||
break;
|
||||
}
|
||||
start = ids.get(ids.size() - 1);
|
||||
} catch (Exception ex) {
|
||||
throw DataXException.asDataXException(GdbReaderErrorCode.FAIL_FETCH_IDS, ex.getMessage());
|
||||
}
|
||||
|
||||
// send range fetch async
|
||||
int count = ids.size();
|
||||
List<ResultSet> resultSets = new LinkedList<>();
|
||||
for (int pos = 0; pos < count; pos += fetchBatchSize) {
|
||||
int rangeSize = Math.min(fetchBatchSize, count - pos);
|
||||
String endId = ids.get(pos + rangeSize - 1);
|
||||
String beginId = ids.get(pos);
|
||||
|
||||
List<String> propNames = rule.isHasProperty() ? rule.getPropertyNames() : null;
|
||||
try {
|
||||
resultSets.add(graph.fetchElementsAsync(fetchLabel, beginId, endId, propNames));
|
||||
} catch (Exception ex) {
|
||||
// just print error logs and continues
|
||||
LOG.error("failed to request label: {}, start: {}, end: {}, e: {}", fetchLabel, beginId, endId, ex);
|
||||
}
|
||||
}
|
||||
|
||||
// get range fetch dsl results
|
||||
resultSets.forEach(results -> {
|
||||
try {
|
||||
List<GdbElement> elements = graph.getElement(results);
|
||||
elements.forEach(element -> {
|
||||
Record record = recordSender.createRecord();
|
||||
DefaultGdbMapper.getMapper(rule).accept(element, record);
|
||||
recordSender.sendToWriter(record);
|
||||
});
|
||||
recordSender.flush();
|
||||
} catch (Exception ex) {
|
||||
LOG.error("failed to send records e {}", ex);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void post() {
|
||||
/**
|
||||
* 注意：此方法每个 Task 都会执行一次。
|
||||
* 最佳实践：如果 Task 中有需要进行数据同步之后的后续处理，可以在此处完成。
|
||||
*/
|
||||
}
|
||||
|
||||
@Override
|
||||
public void destroy() {
|
||||
/**
|
||||
* 注意：此方法每个 Task 都会执行一次。
|
||||
* 最佳实践：通常配合Task 中的 post() 方法一起完成 Task 的资源释放。
|
||||
*/
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
-39
@@ -1,39 +0,0 @@
|
||||
package com.alibaba.datax.plugin.reader.gdbreader;
|
||||
|
||||
import com.alibaba.datax.common.spi.ErrorCode;
|
||||
|
||||
public enum GdbReaderErrorCode implements ErrorCode {
|
||||
/**
|
||||
*
|
||||
*/
|
||||
BAD_CONFIG_VALUE("GdbReader-00", "The value you configured is invalid."),
|
||||
FAIL_CLIENT_CONNECT("GdbReader-02", "GDB connection is abnormal."),
|
||||
UNSUPPORTED_TYPE("GdbReader-03", "Unsupported data type conversion."),
|
||||
FAIL_FETCH_LABELS("GdbReader-04", "Error pulling all labels, it is recommended to configure the specified label pull."),
|
||||
FAIL_FETCH_IDS("GdbReader-05", "Pull range id error."),
|
||||
;
|
||||
|
||||
private final String code;
|
||||
private final String description;
|
||||
|
||||
private GdbReaderErrorCode(String code, String description) {
|
||||
this.code = code;
|
||||
this.description = description;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getCode() {
|
||||
return this.code;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getDescription() {
|
||||
return this.description;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return String.format("Code:[%s], Description:[%s]. ", this.code,
|
||||
this.description);
|
||||
}
|
||||
}
|
||||
@@ -1,86 +0,0 @@
|
||||
package com.alibaba.datax.plugin.reader.gdbreader;
|
||||
|
||||
public final class Key {
|
||||
|
||||
/**
|
||||
* 此处声明插件用到的需要插件使用者提供的配置项
|
||||
*/
|
||||
public final static String HOST = "host";
|
||||
public final static String PORT = "port";
|
||||
public final static String USERNAME = "username";
|
||||
public static final String PASSWORD = "password";
|
||||
|
||||
public static final String LABEL = "labels";
|
||||
public static final String EXPORT_TYPE = "labelType";
|
||||
|
||||
public static final String RANGE_SPLIT_SIZE = "RangeSplitSize";
|
||||
public static final String FETCH_BATCH_SIZE = "fetchBatchSize";
|
||||
|
||||
public static final String COLUMN = "column";
|
||||
public static final String COLUMN_NAME = "name";
|
||||
public static final String COLUMN_TYPE = "type";
|
||||
public static final String COLUMN_NODE_TYPE = "columnType";
|
||||
|
||||
public enum ExportType {
|
||||
/**
|
||||
* Import vertices
|
||||
*/
|
||||
VERTEX,
|
||||
/**
|
||||
* Import edges
|
||||
*/
|
||||
EDGE
|
||||
}
|
||||
|
||||
public enum ColumnType {
|
||||
/**
|
||||
* vertex or edge id
|
||||
*/
|
||||
primaryKey,
|
||||
|
||||
/**
|
||||
* vertex or edge label
|
||||
*/
|
||||
primaryLabel,
|
||||
|
||||
/**
|
||||
* vertex property
|
||||
*/
|
||||
vertexProperty,
|
||||
|
||||
/**
|
||||
* collects all vertex property to Json list
|
||||
*/
|
||||
vertexJsonProperty,
|
||||
|
||||
/**
|
||||
* start vertex id of edge
|
||||
*/
|
||||
srcPrimaryKey,
|
||||
|
||||
/**
|
||||
* start vertex label of edge
|
||||
*/
|
||||
srcPrimaryLabel,
|
||||
|
||||
/**
|
||||
* end vertex id of edge
|
||||
*/
|
||||
dstPrimaryKey,
|
||||
|
||||
/**
|
||||
* end vertex label of edge
|
||||
*/
|
||||
dstPrimaryLabel,
|
||||
|
||||
/**
|
||||
* edge property
|
||||
*/
|
||||
edgeProperty,
|
||||
|
||||
/**
|
||||
* collects all edge property to Json list
|
||||
*/
|
||||
edgeJsonProperty,
|
||||
}
|
||||
}
|
||||
-150
@@ -1,150 +0,0 @@
|
||||
/*
|
||||
* (C) 2019-present Alibaba Group Holding Limited.
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify
|
||||
* it under the terms of the GNU General Public License version 2 as
|
||||
* published by the Free Software Foundation.
|
||||
*/
|
||||
package com.alibaba.datax.plugin.reader.gdbreader.mapping;
|
||||
|
||||
import com.alibaba.datax.common.element.Record;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.model.GdbElement;
|
||||
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceProperty;
|
||||
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceVertexProperty;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.function.BiConsumer;
|
||||
import java.util.function.Function;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
/**
|
||||
* @author : Liu Jianping
|
||||
* @date : 2019/9/6
|
||||
*/
|
||||
|
||||
public class DefaultGdbMapper {
|
||||
|
||||
public static BiConsumer<GdbElement, Record> getMapper(MappingRule rule) {
|
||||
return (gdbElement, record) -> rule.getColumns().forEach(columnMappingRule -> {
|
||||
Object value = null;
|
||||
ValueType type = columnMappingRule.getValueType();
|
||||
String name = columnMappingRule.getName();
|
||||
Map<String, Object> props = gdbElement.getProperties();
|
||||
|
||||
switch (columnMappingRule.getColumnType()) {
|
||||
case dstPrimaryKey:
|
||||
value = gdbElement.getTo();
|
||||
break;
|
||||
case srcPrimaryKey:
|
||||
value = gdbElement.getFrom();
|
||||
break;
|
||||
case primaryKey:
|
||||
value = gdbElement.getId();
|
||||
break;
|
||||
case primaryLabel:
|
||||
value = gdbElement.getLabel();
|
||||
break;
|
||||
case dstPrimaryLabel:
|
||||
value = gdbElement.getToLabel();
|
||||
break;
|
||||
case srcPrimaryLabel:
|
||||
value = gdbElement.getFromLabel();
|
||||
break;
|
||||
case vertexProperty:
|
||||
value = forVertexOnePropertyValue().apply(props.get(name));
|
||||
break;
|
||||
case edgeProperty:
|
||||
value = forEdgePropertyValue().apply(props.get(name));
|
||||
break;
|
||||
case edgeJsonProperty:
|
||||
value = forEdgeJsonProperties().apply(props);
|
||||
break;
|
||||
case vertexJsonProperty:
|
||||
value = forVertexJsonProperties().apply(props);
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
record.addColumn(type.applyObject(value));
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* parser ReferenceProperty value for edge
|
||||
*
|
||||
* @return property value
|
||||
*/
|
||||
private static Function<Object, Object> forEdgePropertyValue() {
|
||||
return prop -> {
|
||||
if (prop instanceof ReferenceProperty) {
|
||||
return ((ReferenceProperty) prop).value();
|
||||
}
|
||||
return null;
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* parser ReferenceVertexProperty value for vertex
|
||||
*
|
||||
* @return the first property value in list
|
||||
*/
|
||||
private static Function<Object, Object> forVertexOnePropertyValue() {
|
||||
return props -> {
|
||||
if (props instanceof List<?>) {
|
||||
// get the first one property if more than one
|
||||
Object o = ((List) props).get(0);
|
||||
if (o instanceof ReferenceVertexProperty) {
|
||||
return ((ReferenceVertexProperty) o).value();
|
||||
}
|
||||
}
|
||||
return null;
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* parser all edge properties to json string
|
||||
*
|
||||
* @return json string
|
||||
*/
|
||||
private static Function<Map<String, Object>, String> forEdgeJsonProperties() {
|
||||
return props -> "{\"properties\":[" +
|
||||
props.entrySet().stream().filter(p -> p.getValue() instanceof ReferenceProperty)
|
||||
.map(p -> "{\"k\":\"" + ((ReferenceProperty) p.getValue()).key() + "\"," +
|
||||
"\"t\":\"" + ((ReferenceProperty) p.getValue()).value().getClass().getSimpleName().toLowerCase() + "\"," +
|
||||
"\"v\":\"" + String.valueOf(((ReferenceProperty) p.getValue()).value()) + "\"}")
|
||||
.collect(Collectors.joining(",")) +
|
||||
"]}";
|
||||
}
|
||||
|
||||
/**
|
||||
* parser all vertex properties to json string, include set-property
|
||||
*
|
||||
* @return json string
|
||||
*/
|
||||
private static Function<Map<String, Object>, String> forVertexJsonProperties() {
|
||||
return props -> "{\"properties\":[" +
|
||||
props.entrySet().stream().filter(p -> p.getValue() instanceof List<?>)
|
||||
.map(p -> forVertexPropertyStr().apply((List<?>) p.getValue()))
|
||||
.collect(Collectors.joining(",")) +
|
||||
"]}";
|
||||
}
|
||||
|
||||
/**
|
||||
* parser one vertex property to json string item, set 'cardinality'
|
||||
*
|
||||
* @return json string item
|
||||
*/
|
||||
private static Function<List<?>, String> forVertexPropertyStr() {
|
||||
return vp -> {
|
||||
final String setFlag = vp.size() > 1 ? "set" : "single";
|
||||
return vp.stream().filter(p -> p instanceof ReferenceVertexProperty)
|
||||
.map(p -> "{\"k\":\"" + ((ReferenceVertexProperty) p).key() + "\"," +
|
||||
"\"t\":\"" + ((ReferenceVertexProperty) p).value().getClass().getSimpleName().toLowerCase() + "\"," +
|
||||
"\"v\":\"" + String.valueOf(((ReferenceVertexProperty) p).value()) + "\"," +
|
||||
"\"c\":\"" + setFlag + "\"}")
|
||||
.collect(Collectors.joining(","));
|
||||
};
|
||||
}
|
||||
}
|
||||
-79
@@ -1,79 +0,0 @@
|
||||
/*
|
||||
* (C) 2019-present Alibaba Group Holding Limited.
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify
|
||||
* it under the terms of the GNU General Public License version 2 as
|
||||
* published by the Free Software Foundation.
|
||||
*/
|
||||
package com.alibaba.datax.plugin.reader.gdbreader.mapping;
|
||||
|
||||
import com.alibaba.datax.common.exception.DataXException;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.GdbReaderErrorCode;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.Key.ColumnType;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.Key.ExportType;
|
||||
import lombok.Data;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* @author : Liu Jianping
|
||||
* @date : 2019/9/6
|
||||
*/
|
||||
|
||||
@Data
|
||||
public class MappingRule {
|
||||
private boolean hasRelation = false;
|
||||
private boolean hasProperty = false;
|
||||
private ExportType type = ExportType.VERTEX;
|
||||
|
||||
/**
|
||||
* property names for property key-value
|
||||
*/
|
||||
private List<String> propertyNames = new ArrayList<>();
|
||||
|
||||
private List<ColumnMappingRule> columns = new ArrayList<>();
|
||||
|
||||
void addColumn(ColumnType columnType, ValueType type, String name) {
|
||||
ColumnMappingRule rule = new ColumnMappingRule();
|
||||
rule.setColumnType(columnType);
|
||||
rule.setName(name);
|
||||
rule.setValueType(type);
|
||||
|
||||
if (columnType == ColumnType.vertexProperty || columnType == ColumnType.edgeProperty) {
|
||||
propertyNames.add(name);
|
||||
hasProperty = true;
|
||||
}
|
||||
|
||||
boolean hasTo = columnType == ColumnType.dstPrimaryKey || columnType == ColumnType.dstPrimaryLabel;
|
||||
boolean hasFrom = columnType == ColumnType.srcPrimaryKey || columnType == ColumnType.srcPrimaryLabel;
|
||||
if (hasTo || hasFrom) {
|
||||
hasRelation = true;
|
||||
}
|
||||
|
||||
columns.add(rule);
|
||||
}
|
||||
|
||||
void addJsonColumn(ColumnType columnType) {
|
||||
ColumnMappingRule rule = new ColumnMappingRule();
|
||||
rule.setColumnType(columnType);
|
||||
rule.setName("json");
|
||||
rule.setValueType(ValueType.STRING);
|
||||
|
||||
if (!propertyNames.isEmpty()) {
|
||||
throw DataXException.asDataXException(GdbReaderErrorCode.BAD_CONFIG_VALUE, "JsonProperties should be only property");
|
||||
}
|
||||
|
||||
columns.add(rule);
|
||||
hasProperty = true;
|
||||
}
|
||||
|
||||
@Data
|
||||
protected static class ColumnMappingRule {
|
||||
private String name = null;
|
||||
|
||||
private ValueType valueType = null;
|
||||
|
||||
private ColumnType columnType = null;
|
||||
}
|
||||
}
|
||||
-76
@@ -1,76 +0,0 @@
|
||||
/*
|
||||
* (C) 2019-present Alibaba Group Holding Limited.
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify
|
||||
* it under the terms of the GNU General Public License version 2 as
|
||||
* published by the Free Software Foundation.
|
||||
*/
|
||||
package com.alibaba.datax.plugin.reader.gdbreader.mapping;
|
||||
|
||||
import com.alibaba.datax.common.exception.DataXException;
|
||||
import com.alibaba.datax.common.util.Configuration;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.GdbReaderErrorCode;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.Key;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.Key.ColumnType;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.Key.ExportType;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.util.ConfigHelper;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* @author : Liu Jianping
|
||||
* @date : 2019/9/20
|
||||
*/
|
||||
|
||||
public class MappingRuleFactory {
|
||||
private static final MappingRuleFactory instance = new MappingRuleFactory();
|
||||
|
||||
public static MappingRuleFactory getInstance() {
|
||||
return instance;
|
||||
}
|
||||
|
||||
public MappingRule create(Configuration config, ExportType exportType) {
|
||||
MappingRule rule = new MappingRule();
|
||||
|
||||
rule.setType(exportType);
|
||||
List<Configuration> configurationList = config.getListConfiguration(Key.COLUMN);
|
||||
for (Configuration column : configurationList) {
|
||||
ColumnType columnType;
|
||||
try {
|
||||
columnType = ColumnType.valueOf(column.getString(Key.COLUMN_NODE_TYPE));
|
||||
} catch (NullPointerException | IllegalArgumentException e) {
|
||||
throw DataXException.asDataXException(GdbReaderErrorCode.BAD_CONFIG_VALUE, Key.COLUMN_NODE_TYPE);
|
||||
}
|
||||
|
||||
if (exportType == ExportType.VERTEX) {
|
||||
// only id/label/property column allow when vertex
|
||||
ConfigHelper.assertConfig(Key.COLUMN_NODE_TYPE, () ->
|
||||
columnType == ColumnType.primaryKey || columnType == ColumnType.primaryLabel
|
||||
|| columnType == ColumnType.vertexProperty || columnType == ColumnType.vertexJsonProperty);
|
||||
} else if (exportType == ExportType.EDGE) {
|
||||
// edge
|
||||
ConfigHelper.assertConfig(Key.COLUMN_NODE_TYPE, () ->
|
||||
columnType == ColumnType.primaryKey || columnType == ColumnType.primaryLabel
|
||||
|| columnType == ColumnType.srcPrimaryKey || columnType == ColumnType.srcPrimaryLabel
|
||||
|| columnType == ColumnType.dstPrimaryKey || columnType == ColumnType.dstPrimaryLabel
|
||||
|| columnType == ColumnType.edgeProperty || columnType == ColumnType.edgeJsonProperty);
|
||||
}
|
||||
|
||||
if (columnType == ColumnType.edgeProperty || columnType == ColumnType.vertexProperty) {
|
||||
String name = column.getString(Key.COLUMN_NAME);
|
||||
ValueType propType = ValueType.fromShortName(column.getString(Key.COLUMN_TYPE));
|
||||
|
||||
ConfigHelper.assertConfig(Key.COLUMN_NAME, () -> name != null);
|
||||
if (propType == null) {
|
||||
throw DataXException.asDataXException(GdbReaderErrorCode.UNSUPPORTED_TYPE, Key.COLUMN_TYPE);
|
||||
}
|
||||
rule.addColumn(columnType, propType, name);
|
||||
} else if (columnType == ColumnType.vertexJsonProperty || columnType == ColumnType.edgeJsonProperty) {
|
||||
rule.addJsonColumn(columnType);
|
||||
} else {
|
||||
rule.addColumn(columnType, ValueType.STRING, null);
|
||||
}
|
||||
}
|
||||
return rule;
|
||||
}
|
||||
}
|
||||
-128
@@ -1,128 +0,0 @@
|
||||
/*
|
||||
* (C) 2019-present Alibaba Group Holding Limited.
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify
|
||||
* it under the terms of the GNU General Public License version 2 as
|
||||
* published by the Free Software Foundation.
|
||||
*/
|
||||
package com.alibaba.datax.plugin.reader.gdbreader.mapping;
|
||||
|
||||
import com.alibaba.datax.common.element.BoolColumn;
|
||||
import com.alibaba.datax.common.element.Column;
|
||||
import com.alibaba.datax.common.element.DoubleColumn;
|
||||
import com.alibaba.datax.common.element.LongColumn;
|
||||
import com.alibaba.datax.common.element.StringColumn;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
import java.util.function.Function;
|
||||
|
||||
/**
|
||||
* @author : Liu Jianping
|
||||
* @date : 2019/9/6
|
||||
*/
|
||||
|
||||
public enum ValueType {
|
||||
/**
|
||||
* transfer gdb element object value to DataX Column data
|
||||
* <p>
|
||||
* int, long -> LongColumn
|
||||
* float, double -> DoubleColumn
|
||||
* bool -> BooleanColumn
|
||||
* string -> StringColumn
|
||||
*/
|
||||
INT(Integer.class, "int", ValueTypeHolder::longColumnMapper),
|
||||
INTEGER(Integer.class, "integer", ValueTypeHolder::longColumnMapper),
|
||||
LONG(Long.class, "long", ValueTypeHolder::longColumnMapper),
|
||||
DOUBLE(Double.class, "double", ValueTypeHolder::doubleColumnMapper),
|
||||
FLOAT(Float.class, "float", ValueTypeHolder::doubleColumnMapper),
|
||||
BOOLEAN(Boolean.class, "boolean", ValueTypeHolder::boolColumnMapper),
|
||||
STRING(String.class, "string", ValueTypeHolder::stringColumnMapper),
|
||||
;
|
||||
|
||||
private Class<?> type = null;
|
||||
private String shortName = null;
|
||||
private Function<Object, Column> columnFunc = null;
|
||||
|
||||
ValueType(Class<?> type, String name, Function<Object, Column> columnFunc) {
|
||||
this.type = type;
|
||||
this.shortName = name;
|
||||
this.columnFunc = columnFunc;
|
||||
|
||||
ValueTypeHolder.shortName2type.put(shortName, this);
|
||||
}
|
||||
|
||||
public static ValueType fromShortName(String name) {
|
||||
return ValueTypeHolder.shortName2type.get(name);
|
||||
}
|
||||
|
||||
public Column applyObject(Object value) {
|
||||
if (value == null) {
|
||||
return null;
|
||||
}
|
||||
return columnFunc.apply(value);
|
||||
}
|
||||
|
||||
private static class ValueTypeHolder {
|
||||
private static Map<String, ValueType> shortName2type = new HashMap<>();
|
||||
|
||||
private static LongColumn longColumnMapper(Object o) {
|
||||
long v;
|
||||
if (o instanceof Integer) {
|
||||
v = (int) o;
|
||||
} else if (o instanceof Long) {
|
||||
v = (long) o;
|
||||
} else if (o instanceof String) {
|
||||
v = Long.valueOf((String) o);
|
||||
} else {
|
||||
throw new RuntimeException("Failed to cast " + o.getClass() + " to Long");
|
||||
}
|
||||
|
||||
return new LongColumn(v);
|
||||
}
|
||||
|
||||
private static DoubleColumn doubleColumnMapper(Object o) {
|
||||
double v;
|
||||
if (o instanceof Integer) {
|
||||
v = (double) (int) o;
|
||||
} else if (o instanceof Long) {
|
||||
v = (double) (long) o;
|
||||
} else if (o instanceof Float) {
|
||||
v = (double) (float) o;
|
||||
} else if (o instanceof Double) {
|
||||
v = (double) o;
|
||||
} else if (o instanceof String) {
|
||||
v = Double.valueOf((String) o);
|
||||
} else {
|
||||
throw new RuntimeException("Failed to cast " + o.getClass() + " to Double");
|
||||
}
|
||||
|
||||
return new DoubleColumn(v);
|
||||
}
|
||||
|
||||
private static BoolColumn boolColumnMapper(Object o) {
|
||||
boolean v;
|
||||
if (o instanceof Integer) {
|
||||
v = ((int) o != 0);
|
||||
} else if (o instanceof Long) {
|
||||
v = ((long) o != 0);
|
||||
} else if (o instanceof Boolean) {
|
||||
v = (boolean) o;
|
||||
} else if (o instanceof String) {
|
||||
v = Boolean.valueOf((String) o);
|
||||
} else {
|
||||
throw new RuntimeException("Failed to cast " + o.getClass() + " to Boolean");
|
||||
}
|
||||
|
||||
return new BoolColumn(v);
|
||||
}
|
||||
|
||||
private static StringColumn stringColumnMapper(Object o) {
|
||||
if (o instanceof String) {
|
||||
return new StringColumn((String) o);
|
||||
} else {
|
||||
return new StringColumn(String.valueOf(o));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
-89
@@ -1,89 +0,0 @@
|
||||
/*
|
||||
* (C) 2019-present Alibaba Group Holding Limited.
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify
|
||||
* it under the terms of the GNU General Public License version 2 as
|
||||
* published by the Free Software Foundation.
|
||||
*/
|
||||
package com.alibaba.datax.plugin.reader.gdbreader.model;
|
||||
|
||||
import com.alibaba.datax.common.util.Configuration;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.Key;
|
||||
import org.apache.tinkerpop.gremlin.driver.Client;
|
||||
import org.apache.tinkerpop.gremlin.driver.Cluster;
|
||||
import org.apache.tinkerpop.gremlin.driver.RequestOptions;
|
||||
import org.apache.tinkerpop.gremlin.driver.Result;
|
||||
import org.apache.tinkerpop.gremlin.driver.ResultSet;
|
||||
import org.apache.tinkerpop.gremlin.driver.ser.Serializers;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* @author : Liu Jianping
|
||||
* @date : 2019/9/6
|
||||
*/
|
||||
|
||||
public abstract class AbstractGdbGraph implements GdbGraph {
|
||||
final static int DEFAULT_TIMEOUT = 30000;
|
||||
private static final Logger log = LoggerFactory.getLogger(AbstractGdbGraph.class);
|
||||
private Client client;
|
||||
|
||||
AbstractGdbGraph() {
|
||||
}
|
||||
|
||||
AbstractGdbGraph(Configuration config) {
|
||||
log.info("init graphdb client");
|
||||
String host = config.getString(Key.HOST);
|
||||
int port = config.getInt(Key.PORT);
|
||||
String username = config.getString(Key.USERNAME);
|
||||
String password = config.getString(Key.PASSWORD);
|
||||
|
||||
try {
|
||||
Cluster cluster = Cluster.build(host).port(port).credentials(username, password)
|
||||
.serializer(Serializers.GRAPHBINARY_V1D0)
|
||||
.maxContentLength(1024 * 1024)
|
||||
.resultIterationBatchSize(64)
|
||||
.create();
|
||||
client = cluster.connect().init();
|
||||
|
||||
warmClient();
|
||||
} catch (RuntimeException e) {
|
||||
log.error("Failed to connect to GDB {}:{}, due to {}", host, port, e);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
protected List<Result> runInternal(String dsl, Map<String, Object> params) throws Exception {
|
||||
return runInternalAsync(dsl, params).all().get(DEFAULT_TIMEOUT + 1000, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
protected ResultSet runInternalAsync(String dsl, Map<String, Object> params) throws Exception {
|
||||
RequestOptions.Builder options = RequestOptions.build().timeout(DEFAULT_TIMEOUT);
|
||||
if (params != null && !params.isEmpty()) {
|
||||
params.forEach(options::addParameter);
|
||||
}
|
||||
return client.submitAsync(dsl, options.create()).get(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
private void warmClient() {
|
||||
try {
|
||||
runInternal("g.V('test')", null);
|
||||
log.info("warm graphdb client over");
|
||||
} catch (Exception e) {
|
||||
log.error("warmClient error");
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws Exception {
|
||||
if (client != null) {
|
||||
log.info("close graphdb client");
|
||||
client.close();
|
||||
}
|
||||
}
|
||||
}
|
||||
-39
@@ -1,39 +0,0 @@
|
||||
/*
|
||||
* (C) 2019-present Alibaba Group Holding Limited.
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify
|
||||
* it under the terms of the GNU General Public License version 2 as
|
||||
* published by the Free Software Foundation.
|
||||
*/
|
||||
package com.alibaba.datax.plugin.reader.gdbreader.model;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* @author : Liu Jianping
|
||||
* @date : 2019/9/6
|
||||
*/
|
||||
|
||||
@Data
|
||||
public class GdbElement {
|
||||
String id = null;
|
||||
String label = null;
|
||||
String to = null;
|
||||
String from = null;
|
||||
String toLabel = null;
|
||||
String fromLabel = null;
|
||||
|
||||
Map<String, Object> properties = new HashMap<>();
|
||||
|
||||
public GdbElement() {
|
||||
}
|
||||
|
||||
public GdbElement(String id, String label) {
|
||||
this.id = id;
|
||||
this.label = label;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,65 +0,0 @@
|
||||
/*
|
||||
* (C) 2019-present Alibaba Group Holding Limited.
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify
|
||||
* it under the terms of the GNU General Public License version 2 as
|
||||
* published by the Free Software Foundation.
|
||||
*/
|
||||
package com.alibaba.datax.plugin.reader.gdbreader.model;
|
||||
|
||||
import org.apache.tinkerpop.gremlin.driver.ResultSet;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* @author : Liu Jianping
|
||||
* @date : 2019/9/6
|
||||
*/
|
||||
|
||||
public interface GdbGraph extends AutoCloseable {
|
||||
|
||||
/**
|
||||
* Get All labels of GraphDB
|
||||
*
|
||||
* @return labels map included numbers
|
||||
*/
|
||||
Map<String, Long> getLabels();
|
||||
|
||||
/**
|
||||
* Get the Ids list of special 'label', size up to 'limit'
|
||||
*
|
||||
* @param label is Label of Vertex or Edge
|
||||
* @param start of Ids range to get
|
||||
* @param limit size of Ids list
|
||||
* @return Ids list
|
||||
*/
|
||||
List<String> fetchIds(String label, String start, long limit);
|
||||
|
||||
/**
|
||||
* Fetch element in async mode, just send query dsl to server
|
||||
*
|
||||
* @param label node label to filter
|
||||
* @param start range begin(included)
|
||||
* @param end range end(included)
|
||||
* @param propNames propKey list to fetch
|
||||
* @return future to get result later
|
||||
*/
|
||||
ResultSet fetchElementsAsync(String label, String start, String end, List<String> propNames);
|
||||
|
||||
/**
|
||||
* Get get element from Response @{ResultSet}
|
||||
*
|
||||
* @param results Response of Server
|
||||
* @return element sets
|
||||
*/
|
||||
List<GdbElement> getElement(ResultSet results);
|
||||
|
||||
/**
|
||||
* close graph client
|
||||
*
|
||||
* @throws Exception if fails
|
||||
*/
|
||||
@Override
|
||||
void close() throws Exception;
|
||||
}
|
||||
-192
@@ -1,192 +0,0 @@
|
||||
/*
|
||||
* (C) 2019-present Alibaba Group Holding Limited.
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify
|
||||
* it under the terms of the GNU General Public License version 2 as
|
||||
* published by the Free Software Foundation.
|
||||
*/
|
||||
package com.alibaba.datax.plugin.reader.gdbreader.model;
|
||||
|
||||
import com.alibaba.datax.common.util.Configuration;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.Key.ExportType;
|
||||
import org.apache.tinkerpop.gremlin.driver.Result;
|
||||
import org.apache.tinkerpop.gremlin.driver.ResultSet;
|
||||
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceEdge;
|
||||
import org.apache.tinkerpop.gremlin.structure.util.reference.ReferenceVertex;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* @author : Liu Jianping
|
||||
* @date : 2019/9/6
|
||||
*/
|
||||
|
||||
public class ScriptGdbGraph extends AbstractGdbGraph {
|
||||
private static final Logger log = LoggerFactory.getLogger(ScriptGdbGraph.class);
|
||||
|
||||
private final static String LABEL = "GDB___LABEL";
|
||||
private final static String START_ID = "GDB___ID";
|
||||
private final static String END_ID = "GDB___ID_END";
|
||||
private final static String LIMIT = "GDB___LIMIT";
|
||||
|
||||
private final static String FETCH_VERTEX_IDS_DSL = "g.V().hasLabel(" + LABEL + ").has(id, gt(" + START_ID + ")).limit(" + LIMIT + ").id()";
|
||||
private final static String FETCH_EDGE_IDS_DSL = "g.E().hasLabel(" + LABEL + ").has(id, gt(" + START_ID + ")).limit(" + LIMIT + ").id()";
|
||||
|
||||
private final static String FETCH_VERTEX_LABELS_DSL = "g.V().groupCount().by(label)";
|
||||
private final static String FETCH_EDGE_LABELS_DSL = "g.E().groupCount().by(label)";
|
||||
|
||||
/**
|
||||
* fetch node range [START_ID, END_ID]
|
||||
*/
|
||||
private final static String FETCH_RANGE_VERTEX_DSL = "g.V().hasLabel(" + LABEL + ").has(id, gte(" + START_ID + ")).has(id, lte(" + END_ID + "))";
|
||||
private final static String FETCH_RANGE_EDGE_DSL = "g.E().hasLabel(" + LABEL + ").has(id, gte(" + START_ID + ")).has(id, lte(" + END_ID + "))";
|
||||
private final static String PART_WITH_PROP_DSL = ".as('a').project('node', 'props').by(select('a')).by(select('a').propertyMap(";
|
||||
|
||||
private final ExportType exportType;
|
||||
|
||||
public ScriptGdbGraph(ExportType exportType) {
|
||||
super();
|
||||
this.exportType = exportType;
|
||||
}
|
||||
|
||||
public ScriptGdbGraph(Configuration config, ExportType exportType) {
|
||||
super(config);
|
||||
this.exportType = exportType;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<String> fetchIds(final String label, final String start, long limit) {
|
||||
Map<String, Object> params = new HashMap<String, Object>(3) {{
|
||||
put(LABEL, label);
|
||||
put(START_ID, start);
|
||||
put(LIMIT, limit);
|
||||
}};
|
||||
String fetchDsl = exportType == ExportType.VERTEX ? FETCH_VERTEX_IDS_DSL : FETCH_EDGE_IDS_DSL;
|
||||
|
||||
List<String> ids = new ArrayList<>();
|
||||
try {
|
||||
List<Result> results = runInternal(fetchDsl, params);
|
||||
|
||||
// transfer result to id string
|
||||
results.forEach(id -> ids.add(id.getString()));
|
||||
} catch (Exception e) {
|
||||
log.error("fetch range node failed, label {}, start {}", label, start);
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
return ids;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ResultSet fetchElementsAsync(final String label, final String start, final String end, final List<String> propNames) {
|
||||
Map<String, Object> params = new HashMap<>(3);
|
||||
params.put(LABEL, label);
|
||||
params.put(START_ID, start);
|
||||
params.put(END_ID, end);
|
||||
|
||||
String prefixDsl = exportType == ExportType.VERTEX ? FETCH_RANGE_VERTEX_DSL : FETCH_RANGE_EDGE_DSL;
|
||||
StringBuilder fetchDsl = new StringBuilder(prefixDsl);
|
||||
if (propNames != null) {
|
||||
fetchDsl.append(PART_WITH_PROP_DSL);
|
||||
for (int i = 0; i < propNames.size(); i++) {
|
||||
String propName = "GDB___PK" + String.valueOf(i);
|
||||
params.put(propName, propNames.get(i));
|
||||
|
||||
fetchDsl.append(propName);
|
||||
if (i != propNames.size() - 1) {
|
||||
fetchDsl.append(", ");
|
||||
}
|
||||
}
|
||||
fetchDsl.append("))");
|
||||
}
|
||||
|
||||
try {
|
||||
return runInternalAsync(fetchDsl.toString(), params);
|
||||
} catch (Exception e) {
|
||||
log.error("Failed to fetch range node startId {}, end {} , e {}", start, end, e);
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
@SuppressWarnings("unchecked")
|
||||
public List<GdbElement> getElement(ResultSet results) {
|
||||
List<GdbElement> elements = new LinkedList<>();
|
||||
try {
|
||||
List<Result> resultList = results.all().get(DEFAULT_TIMEOUT + 1000, TimeUnit.MILLISECONDS);
|
||||
|
||||
resultList.forEach(n -> {
|
||||
Object o = n.getObject();
|
||||
GdbElement element = new GdbElement();
|
||||
if (o instanceof Map) {
|
||||
// project response
|
||||
Object node = ((Map) o).get("node");
|
||||
Object props = ((Map) o).get("props");
|
||||
|
||||
mapNodeToElement(node, element);
|
||||
mapPropToElement((Map<String, Object>) props, element);
|
||||
} else {
|
||||
// range node response
|
||||
mapNodeToElement(n.getObject(), element);
|
||||
}
|
||||
if (element.getId() != null) {
|
||||
elements.add(element);
|
||||
}
|
||||
});
|
||||
} catch (Exception e) {
|
||||
log.error("Failed to get node: {}", e);
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
return elements;
|
||||
}
|
||||
|
||||
private void mapNodeToElement(Object node, GdbElement element) {
|
||||
if (node instanceof ReferenceVertex) {
|
||||
ReferenceVertex v = (ReferenceVertex) node;
|
||||
|
||||
element.setId((String) v.id());
|
||||
element.setLabel(v.label());
|
||||
} else if (node instanceof ReferenceEdge) {
|
||||
ReferenceEdge e = (ReferenceEdge) node;
|
||||
|
||||
element.setId((String) e.id());
|
||||
element.setLabel(e.label());
|
||||
element.setTo((String) e.inVertex().id());
|
||||
element.setToLabel(e.inVertex().label());
|
||||
element.setFrom((String) e.outVertex().id());
|
||||
element.setFromLabel(e.outVertex().label());
|
||||
}
|
||||
}
|
||||
|
||||
private void mapPropToElement(Map<String, Object> props, GdbElement element) {
|
||||
element.setProperties(props);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<String, Long> getLabels() {
|
||||
String dsl = exportType == ExportType.VERTEX ? FETCH_VERTEX_LABELS_DSL : FETCH_EDGE_LABELS_DSL;
|
||||
|
||||
try {
|
||||
List<Result> results = runInternal(dsl, null);
|
||||
Map<String, Long> labelMap = new HashMap<>(2);
|
||||
|
||||
Map<?, ?> labels = results.get(0).get(Map.class);
|
||||
labels.forEach((k, v) -> {
|
||||
String label = (String) k;
|
||||
Long count = (Long) v;
|
||||
labelMap.put(label, count);
|
||||
});
|
||||
|
||||
return labelMap;
|
||||
} catch (Exception e) {
|
||||
log.error("Failed to fetch label list, please give special labels and run again, e {}", e);
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
-77
@@ -1,77 +0,0 @@
|
||||
/*
|
||||
* (C) 2019-present Alibaba Group Holding Limited.
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify
|
||||
* it under the terms of the GNU General Public License version 2 as
|
||||
* published by the Free Software Foundation.
|
||||
*/
|
||||
package com.alibaba.datax.plugin.reader.gdbreader.util;
|
||||
|
||||
import com.alibaba.datax.common.exception.DataXException;
|
||||
import com.alibaba.datax.common.util.Configuration;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.GdbReaderErrorCode;
|
||||
import com.alibaba.datax.plugin.reader.gdbreader.Key;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
/**
|
||||
* @author : Liu Jianping
|
||||
* @date : 2019/9/6
|
||||
*/
|
||||
|
||||
public interface ConfigHelper {
|
||||
static void assertConfig(String key, Supplier<Boolean> f) {
|
||||
if (!f.get()) {
|
||||
throw DataXException.asDataXException(GdbReaderErrorCode.BAD_CONFIG_VALUE, key);
|
||||
}
|
||||
}
|
||||
|
||||
static void assertHasContent(Configuration config, String key) {
|
||||
assertConfig(key, () -> StringUtils.isNotBlank(config.getString(key)));
|
||||
}
|
||||
|
||||
static void assertGdbClient(Configuration config) {
|
||||
assertHasContent(config, Key.HOST);
|
||||
assertConfig(Key.PORT, () -> config.getInt(Key.PORT) > 0);
|
||||
|
||||
assertHasContent(config, Key.USERNAME);
|
||||
assertHasContent(config, Key.PASSWORD);
|
||||
}
|
||||
|
||||
static List<String> assertLabels(Configuration config) {
|
||||
Object labels = config.get(Key.LABEL);
|
||||
if (!(labels instanceof List)) {
|
||||
throw DataXException.asDataXException(GdbReaderErrorCode.BAD_CONFIG_VALUE, "labels should be List");
|
||||
}
|
||||
|
||||
List<?> list = (List<?>) labels;
|
||||
List<String> configLabels = new ArrayList<>(0);
|
||||
list.forEach(n -> configLabels.add(String.valueOf(n)));
|
||||
|
||||
return configLabels;
|
||||
}
|
||||
|
||||
static List<Configuration> splitConfig(Configuration config, List<String> labels) {
|
||||
List<Configuration> configs = new ArrayList<>();
|
||||
for (String label : labels) {
|
||||
Configuration conf = config.clone();
|
||||
conf.set(Key.LABEL, label);
|
||||
|
||||
configs.add(conf);
|
||||
}
|
||||
return configs;
|
||||
}
|
||||
|
||||
static Configuration fromClasspath(String name) {
|
||||
try (InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream(name)) {
|
||||
return Configuration.from(is);
|
||||
} catch (IOException e) {
|
||||
throw new IllegalArgumentException("File not found: " + name);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,6 +0,0 @@
|
||||
{
|
||||
"name": "gdbreader",
|
||||
"class": "com.alibaba.datax.plugin.reader.gdbreader.GdbReader",
|
||||
"description": "useScene: prod. mechanism: connect GDB with gremlin-client, execute 'g.V().propertyMap() or g.E().propertyMap()' to get record",
|
||||
"developer": "alibaba"
|
||||
}
|
||||
@@ -1,77 +0,0 @@
|
||||
{
|
||||
"job": {
|
||||
"setting": {
|
||||
"speed": {
|
||||
"channel": 1
|
||||
},
|
||||
"errorLimit": {
|
||||
"record": 1
|
||||
}
|
||||
},
|
||||
|
||||
"content": [
|
||||
{
|
||||
"reader": {
|
||||
"name": "gdbreader",
|
||||
"parameter": {
|
||||
"host": "10.218.145.24",
|
||||
"port": 8182,
|
||||
"username": "***",
|
||||
"password": "***",
|
||||
"labelType": "EDGE",
|
||||
"labels": ["label1", "label2"],
|
||||
"column": [
|
||||
{
|
||||
"name": "id",
|
||||
"type": "string",
|
||||
"columnType": "primaryKey"
|
||||
},
|
||||
{
|
||||
"name": "label",
|
||||
"type": "string",
|
||||
"columnType": "primaryLabel"
|
||||
},
|
||||
{
|
||||
"name": "srcId",
|
||||
"type": "string",
|
||||
"columnType": "srcPrimaryKey"
|
||||
},
|
||||
{
|
||||
"name": "srcLabel",
|
||||
"type": "string",
|
||||
"columnType": "srcPrimaryLabel"
|
||||
},
|
||||
{
|
||||
"name": "dstId",
|
||||
"type": "string",
|
||||
"columnType": "srcPrimaryKey"
|
||||
},
|
||||
{
|
||||
"name": "dstLabel",
|
||||
"type": "string",
|
||||
"columnType": "srcPrimaryLabel"
|
||||
},
|
||||
{
|
||||
"name": "name",
|
||||
"type": "string",
|
||||
"columnType": "edgeProperty"
|
||||
},
|
||||
{
|
||||
"name": "weight",
|
||||
"type": "double",
|
||||
"columnType": "edgeProperty"
|
||||
}
|
||||
]
|
||||
}
|
||||
},
|
||||
|
||||
"writer": {
|
||||
"name": "streamwriter",
|
||||
"parameter": {
|
||||
"print": true
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -41,14 +41,6 @@ GDBWriter通过DataX框架获取Reader生成的协议数据，使用`g.addV/E(GD
|
||||
{
|
||||
"random": "60,64",
|
||||
"type": "string"
|
||||
},
|
||||
{
|
||||
"random": "100,1000",
|
||||
"type": "long"
|
||||
},
|
||||
{
|
||||
"random": "32,48",
|
||||
"type": "string"
|
||||
}
|
||||
],
|
||||
"sliceRecordCount": 1000
|
||||
@@ -78,18 +70,6 @@ GDBWriter通过DataX框架获取Reader生成的协议数据，使用`g.addV/E(GD
|
||||
"name": "vertex_propKey",
|
||||
"value": "${2}",
|
||||
"type": "string",
|
||||
"columnType": "vertexSetProperty"
|
||||
},
|
||||
{
|
||||
"name": "vertex_propKey",
|
||||
"value": "${3}",
|
||||
"type": "long",
|
||||
"columnType": "vertexSetProperty"
|
||||
},
|
||||
{
|
||||
"name": "vertex_propKey2",
|
||||
"value": "${4}",
|
||||
"type": "string",
|
||||
"columnType": "vertexProperty"
|
||||
}
|
||||
]
|
||||
@@ -310,7 +290,6 @@ GDBWriter通过DataX框架获取Reader生成的协议数据，使用`g.addV/E(GD
|
||||
* primaryKey：表示该字段是主键id
|
||||
* 点枚举值：
|
||||
* vertexProperty：labelType为点时，表示该字段是点的普通属性
|
||||
* vertexSetProperty：labelType为点时，表示该字段是点的SET属性，value是SET属性中的一个属性值
|
||||
* vertexJsonProperty：labelType为点时，表示是点json属性，value结构请见备注**json properties示例**，点配置最多只允许出现一个json属性；
|
||||
* 边枚举值：
|
||||
* srcPrimaryKey：labelType为边时，表示该字段是起点主键id
|
||||
@@ -326,14 +305,6 @@ GDBWriter通过DataX框架获取Reader生成的协议数据，使用`g.addV/E(GD
|
||||
> {"k":"age","t":"int","v":"20"},
|
||||
> {"k":"sex","t":"string","v":"male"}
|
||||
> ]}
|
||||
>
|
||||
> # json格式同样支持给点添加SET属性，格式如下
|
||||
> {"properties":[
|
||||
> {"k":"name","t":"string","v":"tom","c":"set"},
|
||||
> {"k":"name","t":"string","v":"jack","c":"set"},
|
||||
> {"k":"age","t":"int","v":"20"},
|
||||
> {"k":"sex","t":"string","v":"male"}
|
||||
> ]}
|
||||
> ```
|
||||
|
||||
## 4 性能报告
|
||||
@@ -396,5 +367,4 @@ DataX压测机器
|
||||
- GDBWriter插件与用户查询DSL使用相同的GDB实例端口，导入时可能会影响查询性能
|
||||
|
||||
## FAQ
|
||||
1. 使用SET属性需要升级GDB实例到`1.0.20`版本及以上。
|
||||
2. 边只支持普通单值属性，不能给边写SET属性数据。
|
||||
无
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
package com.alibaba.datax.plugin.writer.gdbwriter;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.*;
|
||||
import java.util.function.Function;
|
||||
|
||||
import com.alibaba.datax.common.element.Record;
|
||||
import com.alibaba.datax.common.exception.DataXException;
|
||||
import com.alibaba.datax.common.plugin.RecordReceiver;
|
||||
@@ -13,33 +18,24 @@ import com.alibaba.datax.plugin.writer.gdbwriter.mapping.MappingRule;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.mapping.MappingRuleFactory;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.model.GdbElement;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.model.GdbGraph;
|
||||
|
||||
import groovy.lang.Tuple2;
|
||||
import io.netty.util.concurrent.DefaultThreadFactory;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.LinkedBlockingDeque;
|
||||
import java.util.concurrent.ThreadPoolExecutor;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Function;
|
||||
|
||||
public class GdbWriter extends Writer {
|
||||
private static final Logger log = LoggerFactory.getLogger(GdbWriter.class);
|
||||
private static final Logger log = LoggerFactory.getLogger(GdbWriter.class);
|
||||
|
||||
private static Function<Record, GdbElement> mapper = null;
|
||||
private static GdbGraph globalGraph = null;
|
||||
private static boolean session = false;
|
||||
private static Function<Record, GdbElement> mapper = null;
|
||||
private static GdbGraph globalGraph = null;
|
||||
private static boolean session = false;
|
||||
|
||||
/**
|
||||
* Job 中的方法仅执行一次，Task 中方法会由框架启动多个 Task 线程并行执行。
|
||||
* <p/>
|
||||
* 整个 Writer 执行流程是：
|
||||
*
|
||||
* <pre>
|
||||
* Job类init-->prepare-->split
|
||||
*
|
||||
@@ -50,16 +46,17 @@ public class GdbWriter extends Writer {
|
||||
* </pre>
|
||||
*/
|
||||
public static class Job extends Writer.Job {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(Job.class);
|
||||
private static final Logger LOG = LoggerFactory
|
||||
.getLogger(Job.class);
|
||||
|
||||
private Configuration jobConfig = null;
|
||||
|
||||
|
||||
@Override
|
||||
public void init() {
|
||||
LOG.info("GDB datax plugin writer job init begin ...");
|
||||
this.jobConfig = getPluginJobConf();
|
||||
GdbWriterConfig.of(this.jobConfig);
|
||||
LOG.info("GDB datax plugin writer job init end.");
|
||||
LOG.info("GDB datax plugin writer job init begin ...");
|
||||
this.jobConfig = getPluginJobConf();
|
||||
GdbWriterConfig.of(this.jobConfig);
|
||||
LOG.info("GDB datax plugin writer job init end.");
|
||||
|
||||
/**
|
||||
* 注意：此方法仅执行一次。
|
||||
@@ -74,37 +71,37 @@ public class GdbWriter extends Writer {
|
||||
* 注意：此方法仅执行一次。
|
||||
* 最佳实践：如果 Job 中有需要进行数据同步之前的处理，可以在此处完成，如果没有必要则可以直接去掉。
|
||||
*/
|
||||
super.prepare();
|
||||
super.prepare();
|
||||
|
||||
final MappingRule rule = MappingRuleFactory.getInstance().createV2(this.jobConfig);
|
||||
MappingRule rule = MappingRuleFactory.getInstance().createV2(jobConfig);
|
||||
|
||||
mapper = new DefaultGdbMapper(this.jobConfig).getMapper(rule);
|
||||
session = this.jobConfig.getBool(Key.SESSION_STATE, false);
|
||||
mapper = new DefaultGdbMapper().getMapper(rule);
|
||||
session = jobConfig.getBool(Key.SESSION_STATE, false);
|
||||
|
||||
/**
|
||||
* client connect check before task
|
||||
*/
|
||||
try {
|
||||
globalGraph = GdbGraphManager.instance().getGraph(this.jobConfig, false);
|
||||
} catch (final RuntimeException e) {
|
||||
globalGraph = GdbGraphManager.instance().getGraph(jobConfig, false);
|
||||
} catch (RuntimeException e) {
|
||||
throw DataXException.asDataXException(GdbWriterErrorCode.FAIL_CLIENT_CONNECT, e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<Configuration> split(final int mandatoryNumber) {
|
||||
public List<Configuration> split(int mandatoryNumber) {
|
||||
/**
|
||||
* 注意：此方法仅执行一次。
|
||||
* 最佳实践：通常采用工具静态类完成把 Job 配置切分成多个 Task 配置的工作。
|
||||
* 这里的 mandatoryNumber 是强制必须切分的份数。
|
||||
*/
|
||||
LOG.info("split begin...");
|
||||
final List<Configuration> configurationList = new ArrayList<Configuration>();
|
||||
for (int i = 0; i < mandatoryNumber; i++) {
|
||||
configurationList.add(this.jobConfig.clone());
|
||||
}
|
||||
LOG.info("split end...");
|
||||
return configurationList;
|
||||
LOG.info("split begin...");
|
||||
List<Configuration> configurationList = new ArrayList<Configuration>();
|
||||
for (int i = 0; i < mandatoryNumber; i++) {
|
||||
configurationList.add(this.jobConfig.clone());
|
||||
}
|
||||
LOG.info("split end...");
|
||||
return configurationList;
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -130,7 +127,7 @@ public class GdbWriter extends Writer {
|
||||
public static class Task extends Writer.Task {
|
||||
|
||||
private Configuration taskConfig;
|
||||
|
||||
|
||||
private int failed = 0;
|
||||
private int batchRecords;
|
||||
private ExecutorService submitService = null;
|
||||
@@ -142,24 +139,24 @@ public class GdbWriter extends Writer {
|
||||
* 注意：此方法每个 Task 都会执行一次。
|
||||
* 最佳实践：此处通过对 taskConfig 配置的读取，进而初始化一些资源为 startWrite()做准备。
|
||||
*/
|
||||
this.taskConfig = super.getPluginJobConf();
|
||||
this.batchRecords = this.taskConfig.getInt(Key.MAX_RECORDS_IN_BATCH, GdbWriterConfig.DEFAULT_RECORD_NUM_IN_BATCH);
|
||||
this.submitService = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>(),
|
||||
new DefaultThreadFactory("submit-dsl"));
|
||||
this.taskConfig = super.getPluginJobConf();
|
||||
batchRecords = taskConfig.getInt(Key.MAX_RECORDS_IN_BATCH, GdbWriterConfig.DEFAULT_RECORD_NUM_IN_BATCH);
|
||||
submitService = new ThreadPoolExecutor(1, 1, 0L,
|
||||
TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>(), new DefaultThreadFactory("submit-dsl"));
|
||||
|
||||
if (!session) {
|
||||
this.graph = globalGraph;
|
||||
} else {
|
||||
/**
|
||||
* 分批创建session client，由于服务端groovy编译性能的限制
|
||||
*/
|
||||
try {
|
||||
Thread.sleep((getTaskId() / 10) * 10000);
|
||||
} catch (final Exception e) {
|
||||
// ...
|
||||
}
|
||||
this.graph = GdbGraphManager.instance().getGraph(this.taskConfig, session);
|
||||
}
|
||||
if (!session) {
|
||||
graph = globalGraph;
|
||||
} else {
|
||||
/**
|
||||
* 分批创建session client，由于服务端groovy编译性能的限制
|
||||
*/
|
||||
try {
|
||||
Thread.sleep((getTaskId()/10)*10000);
|
||||
} catch (Exception e) {
|
||||
// ...
|
||||
}
|
||||
graph = GdbGraphManager.instance().getGraph(taskConfig, session);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -168,69 +165,64 @@ public class GdbWriter extends Writer {
|
||||
* 注意：此方法每个 Task 都会执行一次。
|
||||
* 最佳实践：如果 Task 中有需要进行数据同步之前的处理，可以在此处完成，如果没有必要则可以直接去掉。
|
||||
*/
|
||||
super.prepare();
|
||||
super.prepare();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void startWrite(final RecordReceiver recordReceiver) {
|
||||
public void startWrite(RecordReceiver recordReceiver) {
|
||||
/**
|
||||
* 注意：此方法每个 Task 都会执行一次。
|
||||
* 最佳实践：此处适当封装确保简洁清晰完成数据写入工作。
|
||||
*/
|
||||
Record r;
|
||||
Future<Boolean> future = null;
|
||||
List<Tuple2<Record, GdbElement>> records = new ArrayList<>(this.batchRecords);
|
||||
Record r;
|
||||
Future<Boolean> future = null;
|
||||
List<Tuple2<Record, GdbElement>> records = new ArrayList<>(batchRecords);
|
||||
|
||||
while ((r = recordReceiver.getFromReader()) != null) {
|
||||
try {
|
||||
records.add(new Tuple2<>(r, mapper.apply(r)));
|
||||
} catch (final Exception ex) {
|
||||
getTaskPluginCollector().collectDirtyRecord(r, ex);
|
||||
continue;
|
||||
}
|
||||
while ((r = recordReceiver.getFromReader()) != null) {
|
||||
records.add(new Tuple2<>(r, mapper.apply(r)));
|
||||
|
||||
if (records.size() >= this.batchRecords) {
|
||||
wait4Submit(future);
|
||||
if (records.size() >= batchRecords) {
|
||||
wait4Submit(future);
|
||||
|
||||
final List<Tuple2<Record, GdbElement>> batch = records;
|
||||
future = this.submitService.submit(() -> batchCommitRecords(batch));
|
||||
records = new ArrayList<>(this.batchRecords);
|
||||
}
|
||||
}
|
||||
final List<Tuple2<Record, GdbElement>> batch = records;
|
||||
future = submitService.submit(() -> batchCommitRecords(batch));
|
||||
records = new ArrayList<>(batchRecords);
|
||||
}
|
||||
}
|
||||
|
||||
wait4Submit(future);
|
||||
if (!records.isEmpty()) {
|
||||
final List<Tuple2<Record, GdbElement>> batch = records;
|
||||
future = this.submitService.submit(() -> batchCommitRecords(batch));
|
||||
wait4Submit(future);
|
||||
}
|
||||
wait4Submit(future);
|
||||
if (!records.isEmpty()) {
|
||||
final List<Tuple2<Record, GdbElement>> batch = records;
|
||||
future = submitService.submit(() -> batchCommitRecords(batch));
|
||||
wait4Submit(future);
|
||||
}
|
||||
}
|
||||
|
||||
private void wait4Submit(final Future<Boolean> future) {
|
||||
if (future == null) {
|
||||
return;
|
||||
}
|
||||
private void wait4Submit(Future<Boolean> future) {
|
||||
if (future == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
future.get();
|
||||
} catch (final Exception e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
try {
|
||||
future.get();
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
|
||||
private boolean batchCommitRecords(final List<Tuple2<Record, GdbElement>> records) {
|
||||
final TaskPluginCollector collector = getTaskPluginCollector();
|
||||
try {
|
||||
final List<Tuple2<Record, Exception>> errors = this.graph.add(records);
|
||||
errors.forEach(t -> collector.collectDirtyRecord(t.getFirst(), t.getSecond()));
|
||||
this.failed += errors.size();
|
||||
} catch (final Exception e) {
|
||||
records.forEach(t -> collector.collectDirtyRecord(t.getFirst(), e));
|
||||
this.failed += records.size();
|
||||
}
|
||||
TaskPluginCollector collector = getTaskPluginCollector();
|
||||
try {
|
||||
List<Tuple2<Record, Exception>> errors = graph.add(records);
|
||||
errors.forEach(t -> collector.collectDirtyRecord(t.getFirst(), t.getSecond()));
|
||||
failed += errors.size();
|
||||
} catch (Exception e) {
|
||||
records.forEach(t -> collector.collectDirtyRecord(t.getFirst(), e));
|
||||
failed += records.size();
|
||||
}
|
||||
|
||||
records.clear();
|
||||
return true;
|
||||
records.clear();
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -239,7 +231,7 @@ public class GdbWriter extends Writer {
|
||||
* 注意：此方法每个 Task 都会执行一次。
|
||||
* 最佳实践：如果 Task 中有需要进行数据同步之后的后续处理，可以在此处完成。
|
||||
*/
|
||||
log.info("Task done, dirty record count - {}", this.failed);
|
||||
log.info("Task done, dirty record count - {}", failed);
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -249,9 +241,9 @@ public class GdbWriter extends Writer {
|
||||
* 最佳实践：通常配合Task 中的 post() 方法一起完成 Task 的资源释放。
|
||||
*/
|
||||
if (session) {
|
||||
this.graph.close();
|
||||
graph.close();
|
||||
}
|
||||
this.submitService.shutdown();
|
||||
submitService.shutdown();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
+2
-1
@@ -27,6 +27,7 @@ public enum GdbWriterErrorCode implements ErrorCode {
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return String.format("Code:[%s], Description:[%s]. ", this.code, this.description);
|
||||
return String.format("Code:[%s], Description:[%s]. ", this.code,
|
||||
this.description);
|
||||
}
|
||||
}
|
||||
@@ -6,164 +6,136 @@ public final class Key {
|
||||
* 此处声明插件用到的需要插件使用者提供的配置项
|
||||
*/
|
||||
|
||||
public final static String HOST = "host";
|
||||
public final static String PORT = "port";
|
||||
public final static String HOST = "host";
|
||||
public final static String PORT = "port";
|
||||
public final static String USERNAME = "username";
|
||||
public static final String PASSWORD = "password";
|
||||
public static final String PASSWORD = "password";
|
||||
|
||||
/**
|
||||
* import type and mode
|
||||
*/
|
||||
public static final String IMPORT_TYPE = "labelType";
|
||||
public static final String UPDATE_MODE = "writeMode";
|
||||
/**
|
||||
* import type and mode
|
||||
*/
|
||||
public static final String IMPORT_TYPE = "labelType";
|
||||
public static final String UPDATE_MODE = "writeMode";
|
||||
|
||||
/**
|
||||
* label prefix issue
|
||||
*/
|
||||
public static final String ID_TRANS_RULE = "idTransRule";
|
||||
public static final String SRC_ID_TRANS_RULE = "srcIdTransRule";
|
||||
public static final String DST_ID_TRANS_RULE = "dstIdTransRule";
|
||||
/**
|
||||
* label prefix issue
|
||||
*/
|
||||
public static final String ID_TRANS_RULE = "idTransRule";
|
||||
public static final String SRC_ID_TRANS_RULE = "srcIdTransRule";
|
||||
public static final String DST_ID_TRANS_RULE = "dstIdTransRule";
|
||||
|
||||
public static final String LABEL = "label";
|
||||
public static final String SRC_LABEL = "srcLabel";
|
||||
public static final String DST_LABEL = "dstLabel";
|
||||
public static final String LABEL = "label";
|
||||
public static final String SRC_LABEL = "srcLabel";
|
||||
public static final String DST_LABEL = "dstLabel";
|
||||
|
||||
public static final String MAPPING = "mapping";
|
||||
public static final String MAPPING = "mapping";
|
||||
|
||||
/**
|
||||
* column define in Gdb
|
||||
*/
|
||||
public static final String COLUMN = "column";
|
||||
public static final String COLUMN_NAME = "name";
|
||||
public static final String COLUMN_VALUE = "value";
|
||||
public static final String COLUMN_TYPE = "type";
|
||||
public static final String COLUMN_NODE_TYPE = "columnType";
|
||||
/**
|
||||
* column define in Gdb
|
||||
*/
|
||||
public static final String COLUMN = "column";
|
||||
public static final String COLUMN_NAME = "name";
|
||||
public static final String COLUMN_VALUE = "value";
|
||||
public static final String COLUMN_TYPE = "type";
|
||||
public static final String COLUMN_NODE_TYPE = "columnType";
|
||||
|
||||
/**
|
||||
* Gdb Vertex/Edge elements
|
||||
*/
|
||||
public static final String ID = "id";
|
||||
public static final String FROM = "from";
|
||||
public static final String TO = "to";
|
||||
public static final String PROPERTIES = "properties";
|
||||
public static final String PROP_KEY = "name";
|
||||
public static final String PROP_VALUE = "value";
|
||||
public static final String PROP_TYPE = "type";
|
||||
/**
|
||||
* Gdb Vertex/Edge elements
|
||||
*/
|
||||
public static final String ID = "id";
|
||||
public static final String FROM = "from";
|
||||
public static final String TO = "to";
|
||||
public static final String PROPERTIES = "properties";
|
||||
public static final String PROP_KEY = "name";
|
||||
public static final String PROP_VALUE = "value";
|
||||
public static final String PROP_TYPE = "type";
|
||||
|
||||
public static final String PROPERTIES_JSON_STR = "propertiesJsonStr";
|
||||
public static final String MAX_PROPERTIES_BATCH_NUM = "maxPropertiesBatchNumber";
|
||||
public static final String PROPERTIES_JSON_STR = "propertiesJsonStr";
|
||||
public static final String MAX_PROPERTIES_BATCH_NUM = "maxPropertiesBatchNumber";
|
||||
|
||||
/**
|
||||
* session less client configure for connect pool
|
||||
*/
|
||||
public static final String MAX_IN_PROCESS_PER_CONNECTION = "maxInProcessPerConnection";
|
||||
public static final String MAX_CONNECTION_POOL_SIZE = "maxConnectionPoolSize";
|
||||
public static final String MAX_SIMULTANEOUS_USAGE_PER_CONNECTION = "maxSimultaneousUsagePerConnection";
|
||||
/**
|
||||
* session less client configure for connect pool
|
||||
*/
|
||||
public static final String MAX_IN_PROCESS_PER_CONNECTION = "maxInProcessPerConnection";
|
||||
public static final String MAX_CONNECTION_POOL_SIZE = "maxConnectionPoolSize";
|
||||
public static final String MAX_SIMULTANEOUS_USAGE_PER_CONNECTION = "maxSimultaneousUsagePerConnection";
|
||||
|
||||
public static final String MAX_RECORDS_IN_BATCH = "maxRecordsInBatch";
|
||||
public static final String SESSION_STATE = "session";
|
||||
public static final String MAX_RECORDS_IN_BATCH = "maxRecordsInBatch";
|
||||
public static final String SESSION_STATE = "session";
|
||||
|
||||
/**
|
||||
* request length limit, include gdb element string length GDB字段长度限制配置，可分别配置各字段的限制，超过限制的记录会当脏数据处理
|
||||
*/
|
||||
public static final String MAX_GDB_STRING_LENGTH = "maxStringLengthLimit";
|
||||
public static final String MAX_GDB_ID_LENGTH = "maxIdStringLengthLimit";
|
||||
public static final String MAX_GDB_LABEL_LENGTH = "maxLabelStringLengthLimit";
|
||||
public static final String MAX_GDB_PROP_KEY_LENGTH = "maxPropKeyStringLengthLimit";
|
||||
public static final String MAX_GDB_PROP_VALUE_LENGTH = "maxPropValueStringLengthLimit";
|
||||
public static enum ImportType {
|
||||
/**
|
||||
* Import vertices
|
||||
*/
|
||||
VERTEX,
|
||||
/**
|
||||
* Import edges
|
||||
*/
|
||||
EDGE;
|
||||
}
|
||||
|
||||
public static enum UpdateMode {
|
||||
/**
|
||||
* Insert new records, fail if exists
|
||||
*/
|
||||
INSERT,
|
||||
/**
|
||||
* Skip this record if exists
|
||||
*/
|
||||
SKIP,
|
||||
/**
|
||||
* Update property of this record if exists
|
||||
*/
|
||||
MERGE;
|
||||
}
|
||||
|
||||
public static final String MAX_GDB_REQUEST_LENGTH = "maxRequestLengthLimit";
|
||||
public static enum ColumnType {
|
||||
/**
|
||||
* vertex or edge id
|
||||
*/
|
||||
primaryKey,
|
||||
|
||||
public static enum ImportType {
|
||||
/**
|
||||
* Import vertices
|
||||
*/
|
||||
VERTEX,
|
||||
/**
|
||||
* Import edges
|
||||
*/
|
||||
EDGE;
|
||||
}
|
||||
/**
|
||||
* vertex property
|
||||
*/
|
||||
vertexProperty,
|
||||
|
||||
public static enum UpdateMode {
|
||||
/**
|
||||
* Insert new records, fail if exists
|
||||
*/
|
||||
INSERT,
|
||||
/**
|
||||
* Skip this record if exists
|
||||
*/
|
||||
SKIP,
|
||||
/**
|
||||
* Update property of this record if exists
|
||||
*/
|
||||
MERGE;
|
||||
}
|
||||
/**
|
||||
* start vertex id of edge
|
||||
*/
|
||||
srcPrimaryKey,
|
||||
|
||||
public static enum ColumnType {
|
||||
/**
|
||||
* vertex or edge id
|
||||
*/
|
||||
primaryKey,
|
||||
/**
|
||||
* end vertex id of edge
|
||||
*/
|
||||
dstPrimaryKey,
|
||||
|
||||
/**
|
||||
* vertex property
|
||||
*/
|
||||
vertexProperty,
|
||||
/**
|
||||
* edge property
|
||||
*/
|
||||
edgeProperty,
|
||||
|
||||
/**
|
||||
* vertex setProperty
|
||||
*/
|
||||
vertexSetProperty,
|
||||
/**
|
||||
* vertex json style property
|
||||
*/
|
||||
vertexJsonProperty,
|
||||
|
||||
/**
|
||||
* start vertex id of edge
|
||||
*/
|
||||
srcPrimaryKey,
|
||||
/**
|
||||
* edge json style property
|
||||
*/
|
||||
edgeJsonProperty
|
||||
}
|
||||
|
||||
/**
|
||||
* end vertex id of edge
|
||||
*/
|
||||
dstPrimaryKey,
|
||||
public static enum IdTransRule {
|
||||
/**
|
||||
* vertex or edge id with 'label' prefix
|
||||
*/
|
||||
labelPrefix,
|
||||
|
||||
/**
|
||||
* edge property
|
||||
*/
|
||||
edgeProperty,
|
||||
|
||||
/**
|
||||
* vertex json style property
|
||||
*/
|
||||
vertexJsonProperty,
|
||||
|
||||
/**
|
||||
* edge json style property
|
||||
*/
|
||||
edgeJsonProperty
|
||||
}
|
||||
|
||||
public static enum IdTransRule {
|
||||
/**
|
||||
* vertex or edge id with 'label' prefix
|
||||
*/
|
||||
labelPrefix,
|
||||
|
||||
/**
|
||||
* vertex or edge id raw
|
||||
*/
|
||||
none
|
||||
}
|
||||
|
||||
public static enum PropertyType {
|
||||
/**
|
||||
* single Vertex Property
|
||||
*/
|
||||
single,
|
||||
|
||||
/**
|
||||
* set Vertex Property
|
||||
*/
|
||||
set
|
||||
}
|
||||
/**
|
||||
* vertex or edge id raw
|
||||
*/
|
||||
none
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
+22
-22
@@ -3,37 +3,37 @@
|
||||
*/
|
||||
package com.alibaba.datax.plugin.writer.gdbwriter.client;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import com.alibaba.datax.common.util.Configuration;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.model.GdbGraph;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.model.ScriptGdbGraph;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* @author jerrywang
|
||||
*
|
||||
*/
|
||||
public class GdbGraphManager implements AutoCloseable {
|
||||
private static final GdbGraphManager INSTANCE = new GdbGraphManager();
|
||||
private static final GdbGraphManager instance = new GdbGraphManager();
|
||||
|
||||
private List<GdbGraph> graphs = new ArrayList<>();
|
||||
|
||||
public static GdbGraphManager instance() {
|
||||
return instance;
|
||||
}
|
||||
|
||||
private List<GdbGraph> graphs = new ArrayList<>();
|
||||
public GdbGraph getGraph(Configuration config, boolean session) {
|
||||
GdbGraph graph = new ScriptGdbGraph(config, session);
|
||||
graphs.add(graph);
|
||||
return graph;
|
||||
}
|
||||
|
||||
public static GdbGraphManager instance() {
|
||||
return INSTANCE;
|
||||
}
|
||||
|
||||
public GdbGraph getGraph(final Configuration config, final boolean session) {
|
||||
final GdbGraph graph = new ScriptGdbGraph(config, session);
|
||||
this.graphs.add(graph);
|
||||
return graph;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
for (final GdbGraph graph : this.graphs) {
|
||||
graph.close();
|
||||
}
|
||||
this.graphs.clear();
|
||||
}
|
||||
@Override
|
||||
public void close() {
|
||||
for(GdbGraph graph : graphs) {
|
||||
graph.close();
|
||||
}
|
||||
graphs.clear();
|
||||
}
|
||||
}
|
||||
|
||||
+22
-26
@@ -3,43 +3,39 @@
|
||||
*/
|
||||
package com.alibaba.datax.plugin.writer.gdbwriter.client;
|
||||
|
||||
import static com.alibaba.datax.plugin.writer.gdbwriter.util.ConfigHelper.assertConfig;
|
||||
import static com.alibaba.datax.plugin.writer.gdbwriter.util.ConfigHelper.assertHasContent;
|
||||
|
||||
import com.alibaba.datax.common.util.Configuration;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.Key;
|
||||
|
||||
import static com.alibaba.datax.plugin.writer.gdbwriter.util.ConfigHelper.*;
|
||||
|
||||
/**
|
||||
* @author jerrywang
|
||||
*
|
||||
*/
|
||||
public class GdbWriterConfig {
|
||||
public static final int DEFAULT_MAX_IN_PROCESS_PER_CONNECTION = 4;
|
||||
public static final int DEFAULT_MAX_CONNECTION_POOL_SIZE = 8;
|
||||
public static final int DEFAULT_MAX_SIMULTANEOUS_USAGE_PER_CONNECTION = 8;
|
||||
public static final int DEFAULT_BATCH_PROPERTY_NUM = 30;
|
||||
public static final int DEFAULT_RECORD_NUM_IN_BATCH = 16;
|
||||
public static final int DEFAULT_MAX_IN_PROCESS_PER_CONNECTION = 4;
|
||||
public static final int DEFAULT_MAX_CONNECTION_POOL_SIZE = 8;
|
||||
public static final int DEFAULT_MAX_SIMULTANEOUS_USAGE_PER_CONNECTION = 8;
|
||||
public static final int DEFAULT_BATCH_PROPERTY_NUM = 30;
|
||||
public static final int DEFAULT_RECORD_NUM_IN_BATCH = 16;
|
||||
|
||||
public static final int MAX_STRING_LENGTH = 10240;
|
||||
public static final int MAX_REQUEST_LENGTH = 65535 - 1000;
|
||||
private Configuration config;
|
||||
|
||||
private Configuration config;
|
||||
private GdbWriterConfig(Configuration config) {
|
||||
this.config = config;
|
||||
|
||||
private GdbWriterConfig(final Configuration config) {
|
||||
this.config = config;
|
||||
validate();
|
||||
}
|
||||
|
||||
validate();
|
||||
}
|
||||
private void validate() {
|
||||
assertHasContent(config, Key.HOST);
|
||||
assertConfig(Key.PORT, () -> config.getInt(Key.PORT) > 0);
|
||||
|
||||
public static GdbWriterConfig of(final Configuration config) {
|
||||
return new GdbWriterConfig(config);
|
||||
}
|
||||
|
||||
private void validate() {
|
||||
assertHasContent(this.config, Key.HOST);
|
||||
assertConfig(Key.PORT, () -> this.config.getInt(Key.PORT) > 0);
|
||||
|
||||
assertHasContent(this.config, Key.USERNAME);
|
||||
assertHasContent(this.config, Key.PASSWORD);
|
||||
}
|
||||
assertHasContent(config, Key.USERNAME);
|
||||
assertHasContent(config, Key.PASSWORD);
|
||||
}
|
||||
|
||||
public static GdbWriterConfig of(Configuration config) {
|
||||
return new GdbWriterConfig(config);
|
||||
}
|
||||
}
|
||||
|
||||
+150
-164
@@ -3,8 +3,6 @@
|
||||
*/
|
||||
package com.alibaba.datax.plugin.writer.gdbwriter.mapping;
|
||||
|
||||
import static com.alibaba.datax.plugin.writer.gdbwriter.Key.ImportType.VERTEX;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
@@ -14,191 +12,179 @@ import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
import com.alibaba.datax.common.element.Record;
|
||||
import com.alibaba.datax.common.util.Configuration;
|
||||
import com.alibaba.fastjson.JSONArray;
|
||||
import com.alibaba.fastjson.JSONObject;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.Key;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.model.GdbEdge;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.model.GdbElement;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.model.GdbVertex;
|
||||
import com.alibaba.fastjson.JSONArray;
|
||||
import com.alibaba.fastjson.JSONObject;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import static com.alibaba.datax.plugin.writer.gdbwriter.Key.ImportType.VERTEX;
|
||||
|
||||
/**
|
||||
* @author jerrywang
|
||||
*
|
||||
*/
|
||||
@Slf4j
|
||||
public class DefaultGdbMapper implements GdbMapper {
|
||||
private static final Pattern STR_DOLLAR_PATTERN = Pattern.compile("\\$\\{(\\d+)}");
|
||||
private static final Pattern NORMAL_DOLLAR_PATTERN = Pattern.compile("^\\$\\{(\\d+)}$");
|
||||
private static final Pattern STR_PATTERN = Pattern.compile("\\$\\{(\\d+)}");
|
||||
private static final Pattern NORMAL_PATTERN = Pattern.compile("^\\$\\{(\\d+)}$");
|
||||
|
||||
private static final Pattern STR_NUM_PATTERN = Pattern.compile("#\\{(\\d+)}");
|
||||
private static final Pattern NORMAL_NUM_PATTERN = Pattern.compile("^#\\{(\\d+)}$");
|
||||
|
||||
public DefaultGdbMapper() {}
|
||||
|
||||
public DefaultGdbMapper(final Configuration config) {
|
||||
MapperConfig.getInstance().updateConfig(config);
|
||||
}
|
||||
|
||||
private static BiConsumer<Record, GdbElement> forElement(final MappingRule rule) {
|
||||
final boolean numPattern = rule.isNumPattern();
|
||||
final List<BiConsumer<Record, GdbElement>> properties = new ArrayList<>();
|
||||
for (final MappingRule.PropertyMappingRule propRule : rule.getProperties()) {
|
||||
final Function<Record, String> keyFunc = forStrColumn(numPattern, propRule.getKey());
|
||||
|
||||
if (propRule.getValueType() == ValueType.STRING) {
|
||||
final Function<Record, String> valueFunc = forStrColumn(numPattern, propRule.getValue());
|
||||
properties.add((r, e) -> {
|
||||
e.addProperty(keyFunc.apply(r), valueFunc.apply(r), propRule.getPType());
|
||||
});
|
||||
} else {
|
||||
final Function<Record, Object> valueFunc =
|
||||
forObjColumn(numPattern, propRule.getValue(), propRule.getValueType());
|
||||
properties.add((r, e) -> {
|
||||
e.addProperty(keyFunc.apply(r), valueFunc.apply(r), propRule.getPType());
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
if (rule.getPropertiesJsonStr() != null) {
|
||||
final Function<Record, String> jsonFunc = forStrColumn(numPattern, rule.getPropertiesJsonStr());
|
||||
properties.add((r, e) -> {
|
||||
final String propertiesStr = jsonFunc.apply(r);
|
||||
final JSONObject root = (JSONObject)JSONObject.parse(propertiesStr);
|
||||
final JSONArray propertiesList = root.getJSONArray("properties");
|
||||
|
||||
for (final Object object : propertiesList) {
|
||||
final JSONObject jsonObject = (JSONObject)object;
|
||||
final String key = jsonObject.getString("k");
|
||||
final String name = jsonObject.getString("v");
|
||||
final String type = jsonObject.getString("t");
|
||||
final String card = jsonObject.getString("c");
|
||||
|
||||
if (key == null || name == null) {
|
||||
continue;
|
||||
}
|
||||
addToProperties(e, key, name, type, card);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
final BiConsumer<Record, GdbElement> ret = (r, e) -> {
|
||||
final String label = forStrColumn(numPattern, rule.getLabel()).apply(r);
|
||||
String id = forStrColumn(numPattern, rule.getId()).apply(r);
|
||||
|
||||
if (rule.getImportType() == Key.ImportType.EDGE) {
|
||||
final String to = forStrColumn(numPattern, rule.getTo()).apply(r);
|
||||
final String from = forStrColumn(numPattern, rule.getFrom()).apply(r);
|
||||
if (to == null || from == null) {
|
||||
log.error("invalid record to: {} , from: {}", to, from);
|
||||
throw new IllegalArgumentException("to or from missed in edge");
|
||||
}
|
||||
((GdbEdge)e).setTo(to);
|
||||
((GdbEdge)e).setFrom(from);
|
||||
|
||||
// generate UUID for edge
|
||||
if (id == null) {
|
||||
id = UUID.randomUUID().toString();
|
||||
}
|
||||
}
|
||||
|
||||
if (id == null || label == null) {
|
||||
log.error("invalid record id: {} , label: {}", id, label);
|
||||
throw new IllegalArgumentException("id or label missed");
|
||||
}
|
||||
|
||||
e.setId(id);
|
||||
e.setLabel(label);
|
||||
|
||||
properties.forEach(p -> p.accept(r, e));
|
||||
@Override
|
||||
public Function<Record, GdbElement> getMapper(MappingRule rule) {
|
||||
return r -> {
|
||||
GdbElement e = (rule.getImportType() == VERTEX) ? new GdbVertex() : new GdbEdge();
|
||||
forElement(rule).accept(r, e);
|
||||
return e;
|
||||
};
|
||||
return ret;
|
||||
}
|
||||
}
|
||||
|
||||
private static Function<Record, Object> forObjColumn(final boolean numPattern, final String rule, final ValueType type) {
|
||||
final Pattern pattern = numPattern ? NORMAL_NUM_PATTERN : NORMAL_DOLLAR_PATTERN;
|
||||
final Matcher m = pattern.matcher(rule);
|
||||
if (m.matches()) {
|
||||
final int index = Integer.valueOf(m.group(1));
|
||||
return r -> type.applyColumn(r.getColumn(index));
|
||||
} else {
|
||||
return r -> type.fromStrFunc(rule);
|
||||
}
|
||||
}
|
||||
private static BiConsumer<Record, GdbElement> forElement(MappingRule rule) {
|
||||
List<BiConsumer<Record, GdbElement>> properties = new ArrayList<>();
|
||||
for (MappingRule.PropertyMappingRule propRule : rule.getProperties()) {
|
||||
Function<Record, String> keyFunc = forStrColumn(propRule.getKey());
|
||||
|
||||
private static Function<Record, String> forStrColumn(final boolean numPattern, final String rule) {
|
||||
final List<BiConsumer<StringBuilder, Record>> list = new ArrayList<>();
|
||||
final Pattern pattern = numPattern ? STR_NUM_PATTERN : STR_DOLLAR_PATTERN;
|
||||
final Matcher m = pattern.matcher(rule);
|
||||
int last = 0;
|
||||
while (m.find()) {
|
||||
final String index = m.group(1);
|
||||
// as simple integer index.
|
||||
final int i = Integer.parseInt(index);
|
||||
if (propRule.getValueType() == ValueType.STRING) {
|
||||
final Function<Record, String> valueFunc = forStrColumn(propRule.getValue());
|
||||
properties.add((r, e) -> {
|
||||
String k = keyFunc.apply(r);
|
||||
String v = valueFunc.apply(r);
|
||||
if (k != null && v != null) {
|
||||
e.getProperties().put(k, v);
|
||||
}
|
||||
});
|
||||
} else {
|
||||
final Function<Record, Object> valueFunc = forObjColumn(propRule.getValue(), propRule.getValueType());
|
||||
properties.add((r, e) -> {
|
||||
String k = keyFunc.apply(r);
|
||||
Object v = valueFunc.apply(r);
|
||||
if (k != null && v != null) {
|
||||
e.getProperties().put(k, v);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
final int tmp = last;
|
||||
final int start = m.start();
|
||||
list.add((sb, record) -> {
|
||||
sb.append(rule.subSequence(tmp, start));
|
||||
if (record.getColumn(i) != null && record.getColumn(i).getByteSize() > 0) {
|
||||
sb.append(record.getColumn(i).asString());
|
||||
}
|
||||
});
|
||||
if (rule.getPropertiesJsonStr() != null) {
|
||||
Function<Record, String> jsonFunc = forStrColumn(rule.getPropertiesJsonStr());
|
||||
properties.add((r, e) -> {
|
||||
String propertiesStr = jsonFunc.apply(r);
|
||||
JSONObject root = (JSONObject)JSONObject.parse(propertiesStr);
|
||||
JSONArray propertiesList = root.getJSONArray("properties");
|
||||
|
||||
last = m.end();
|
||||
}
|
||||
for (Object object : propertiesList) {
|
||||
JSONObject jsonObject = (JSONObject)object;
|
||||
String key = jsonObject.getString("k");
|
||||
String name = jsonObject.getString("v");
|
||||
String type = jsonObject.getString("t");
|
||||
|
||||
final int tmp = last;
|
||||
list.add((sb, record) -> {
|
||||
sb.append(rule.subSequence(tmp, rule.length()));
|
||||
});
|
||||
if (key == null || name == null) {
|
||||
continue;
|
||||
}
|
||||
addToProperties(e, key, name, type);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
return r -> {
|
||||
final StringBuilder sb = new StringBuilder();
|
||||
list.forEach(c -> c.accept(sb, r));
|
||||
final String res = sb.toString();
|
||||
return res.isEmpty() ? null : res;
|
||||
};
|
||||
}
|
||||
BiConsumer<Record, GdbElement> ret = (r, e) -> {
|
||||
String label = forStrColumn(rule.getLabel()).apply(r);
|
||||
String id = forStrColumn(rule.getId()).apply(r);
|
||||
|
||||
private static boolean addToProperties(final GdbElement e, final String key, final String value, final String type, final String card) {
|
||||
final Object pValue;
|
||||
final ValueType valueType = ValueType.fromShortName(type);
|
||||
if (rule.getImportType() == Key.ImportType.EDGE) {
|
||||
String to = forStrColumn(rule.getTo()).apply(r);
|
||||
String from = forStrColumn(rule.getFrom()).apply(r);
|
||||
if (to == null || from == null) {
|
||||
log.error("invalid record to: {} , from: {}", to, from);
|
||||
throw new IllegalArgumentException("to or from missed in edge");
|
||||
}
|
||||
((GdbEdge)e).setTo(to);
|
||||
((GdbEdge)e).setFrom(from);
|
||||
|
||||
if (valueType == ValueType.STRING) {
|
||||
pValue = value;
|
||||
} else if (valueType == ValueType.INT || valueType == ValueType.INTEGER) {
|
||||
pValue = Integer.valueOf(value);
|
||||
} else if (valueType == ValueType.LONG) {
|
||||
pValue = Long.valueOf(value);
|
||||
} else if (valueType == ValueType.DOUBLE) {
|
||||
pValue = Double.valueOf(value);
|
||||
} else if (valueType == ValueType.FLOAT) {
|
||||
pValue = Float.valueOf(value);
|
||||
} else if (valueType == ValueType.BOOLEAN) {
|
||||
pValue = Boolean.valueOf(value);
|
||||
} else {
|
||||
log.error("invalid property key {}, value {}, type {}", key, value, type);
|
||||
return false;
|
||||
}
|
||||
// generate UUID for edge
|
||||
if (id == null) {
|
||||
id = UUID.randomUUID().toString();
|
||||
}
|
||||
}
|
||||
|
||||
// apply vertexSetProperty
|
||||
if (Key.PropertyType.set.name().equals(card) && (e instanceof GdbVertex)) {
|
||||
e.addProperty(key, pValue, Key.PropertyType.set);
|
||||
} else {
|
||||
e.addProperty(key, pValue);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
if (id == null || label == null) {
|
||||
log.error("invalid record id: {} , label: {}", id, label);
|
||||
throw new IllegalArgumentException("id or label missed");
|
||||
}
|
||||
|
||||
@Override
|
||||
public Function<Record, GdbElement> getMapper(final MappingRule rule) {
|
||||
return r -> {
|
||||
final GdbElement e = (rule.getImportType() == VERTEX) ? new GdbVertex() : new GdbEdge();
|
||||
forElement(rule).accept(r, e);
|
||||
return e;
|
||||
};
|
||||
}
|
||||
e.setId(id);
|
||||
e.setLabel(label);
|
||||
|
||||
properties.forEach(p -> p.accept(r, e));
|
||||
};
|
||||
return ret;
|
||||
}
|
||||
|
||||
static Function<Record, Object> forObjColumn(String rule, ValueType type) {
|
||||
Matcher m = NORMAL_PATTERN.matcher(rule);
|
||||
if (m.matches()) {
|
||||
int index = Integer.valueOf(m.group(1));
|
||||
return r -> type.applyColumn(r.getColumn(index));
|
||||
} else {
|
||||
return r -> type.fromStrFunc(rule);
|
||||
}
|
||||
}
|
||||
|
||||
static Function<Record, String> forStrColumn(String rule) {
|
||||
List<BiConsumer<StringBuilder, Record>> list = new ArrayList<>();
|
||||
Matcher m = STR_PATTERN.matcher(rule);
|
||||
int last = 0;
|
||||
while (m.find()) {
|
||||
String index = m.group(1);
|
||||
// as simple integer index.
|
||||
int i = Integer.parseInt(index);
|
||||
|
||||
final int tmp = last;
|
||||
final int start = m.start();
|
||||
list.add((sb, record) -> {
|
||||
sb.append(rule.subSequence(tmp, start));
|
||||
if(record.getColumn(i) != null && record.getColumn(i).getByteSize() > 0) {
|
||||
sb.append(record.getColumn(i).asString());
|
||||
}
|
||||
});
|
||||
|
||||
last = m.end();
|
||||
}
|
||||
|
||||
final int tmp = last;
|
||||
list.add((sb, record) -> {
|
||||
sb.append(rule.subSequence(tmp, rule.length()));
|
||||
});
|
||||
|
||||
return r -> {
|
||||
StringBuilder sb = new StringBuilder();
|
||||
list.forEach(c -> c.accept(sb, r));
|
||||
String res = sb.toString();
|
||||
return res.isEmpty() ? null : res;
|
||||
};
|
||||
}
|
||||
|
||||
static boolean addToProperties(GdbElement e, String key, String value, String type) {
|
||||
ValueType valueType = ValueType.fromShortName(type);
|
||||
|
||||
if(valueType == ValueType.STRING) {
|
||||
e.getProperties().put(key, value);
|
||||
} else if (valueType == ValueType.INT) {
|
||||
e.getProperties().put(key, Integer.valueOf(value));
|
||||
} else if (valueType == ValueType.LONG) {
|
||||
e.getProperties().put(key, Long.valueOf(value));
|
||||
} else if (valueType == ValueType.DOUBLE) {
|
||||
e.getProperties().put(key, Double.valueOf(value));
|
||||
} else if (valueType == ValueType.FLOAT) {
|
||||
e.getProperties().put(key, Float.valueOf(value));
|
||||
} else if (valueType == ValueType.BOOLEAN) {
|
||||
e.getProperties().put(key, Boolean.valueOf(value));
|
||||
} else {
|
||||
log.error("invalid property key {}, value {}, type {}", key, value, type);
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
+1
-1
@@ -13,5 +13,5 @@ import com.alibaba.datax.plugin.writer.gdbwriter.model.GdbElement;
|
||||
*
|
||||
*/
|
||||
public interface GdbMapper {
|
||||
Function<Record, GdbElement> getMapper(MappingRule rule);
|
||||
Function<Record, GdbElement> getMapper(MappingRule rule);
|
||||
}
|
||||
|
||||
-68
@@ -1,68 +0,0 @@
|
||||
/*
|
||||
* (C) 2019-present Alibaba Group Holding Limited.
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public
|
||||
* License version 2 as published by the Free Software Foundation.
|
||||
*/
|
||||
package com.alibaba.datax.plugin.writer.gdbwriter.mapping;
|
||||
|
||||
import com.alibaba.datax.common.util.Configuration;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.Key;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.client.GdbWriterConfig;
|
||||
|
||||
/**
|
||||
* @author : Liu Jianping
|
||||
* @date : 2019/10/15
|
||||
*/
|
||||
|
||||
public class MapperConfig {
|
||||
private static MapperConfig instance = new MapperConfig();
|
||||
private int maxIdLength;
|
||||
private int maxLabelLength;
|
||||
private int maxPropKeyLength;
|
||||
private int maxPropValueLength;
|
||||
|
||||
private MapperConfig() {
|
||||
this.maxIdLength = GdbWriterConfig.MAX_STRING_LENGTH;
|
||||
this.maxLabelLength = GdbWriterConfig.MAX_STRING_LENGTH;
|
||||
this.maxPropKeyLength = GdbWriterConfig.MAX_STRING_LENGTH;
|
||||
this.maxPropValueLength = GdbWriterConfig.MAX_STRING_LENGTH;
|
||||
}
|
||||
|
||||
public static MapperConfig getInstance() {
|
||||
return instance;
|
||||
}
|
||||
|
||||
public void updateConfig(final Configuration config) {
|
||||
final int length = config.getInt(Key.MAX_GDB_STRING_LENGTH, GdbWriterConfig.MAX_STRING_LENGTH);
|
||||
|
||||
Integer sLength = config.getInt(Key.MAX_GDB_ID_LENGTH);
|
||||
this.maxIdLength = sLength == null ? length : sLength;
|
||||
|
||||
sLength = config.getInt(Key.MAX_GDB_LABEL_LENGTH);
|
||||
this.maxLabelLength = sLength == null ? length : sLength;
|
||||
|
||||
sLength = config.getInt(Key.MAX_GDB_PROP_KEY_LENGTH);
|
||||
this.maxPropKeyLength = sLength == null ? length : sLength;
|
||||
|
||||
sLength = config.getInt(Key.MAX_GDB_PROP_VALUE_LENGTH);
|
||||
this.maxPropValueLength = sLength == null ? length : sLength;
|
||||
}
|
||||
|
||||
public int getMaxIdLength() {
|
||||
return this.maxIdLength;
|
||||
}
|
||||
|
||||
public int getMaxLabelLength() {
|
||||
return this.maxLabelLength;
|
||||
}
|
||||
|
||||
public int getMaxPropKeyLength() {
|
||||
return this.maxPropKeyLength;
|
||||
}
|
||||
|
||||
public int getMaxPropValueLength() {
|
||||
return this.maxPropValueLength;
|
||||
}
|
||||
|
||||
}
|
||||
+17
-22
@@ -7,7 +7,6 @@ import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.Key.ImportType;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.Key.PropertyType;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
@@ -17,30 +16,26 @@ import lombok.Data;
|
||||
*/
|
||||
@Data
|
||||
public class MappingRule {
|
||||
private String id = null;
|
||||
private String id = null;
|
||||
|
||||
private String label = null;
|
||||
private String label = null;
|
||||
|
||||
private ImportType importType = null;
|
||||
|
||||
private String from = null;
|
||||
|
||||
private ImportType importType = null;
|
||||
private String to = null;
|
||||
|
||||
private String from = null;
|
||||
private List<PropertyMappingRule> properties = new ArrayList<>();
|
||||
|
||||
private String to = null;
|
||||
private String propertiesJsonStr = null;
|
||||
|
||||
private List<PropertyMappingRule> properties = new ArrayList<>();
|
||||
|
||||
private String propertiesJsonStr = null;
|
||||
|
||||
private boolean numPattern = false;
|
||||
|
||||
@Data
|
||||
public static class PropertyMappingRule {
|
||||
private String key = null;
|
||||
|
||||
private String value = null;
|
||||
|
||||
private ValueType valueType = null;
|
||||
|
||||
private PropertyType pType = PropertyType.single;
|
||||
}
|
||||
@Data
|
||||
public static class PropertyMappingRule {
|
||||
private String key = null;
|
||||
|
||||
private String value = null;
|
||||
|
||||
private ValueType valueType = null;
|
||||
}
|
||||
}
|
||||
|
||||
+116
-155
@@ -3,116 +3,85 @@
|
||||
*/
|
||||
package com.alibaba.datax.plugin.writer.gdbwriter.mapping;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
import com.alibaba.datax.common.exception.DataXException;
|
||||
import com.alibaba.datax.common.util.Configuration;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.GdbWriterErrorCode;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.Key;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.Key.ColumnType;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.Key.IdTransRule;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.Key.ImportType;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.Key.IdTransRule;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.Key.ColumnType;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.mapping.MappingRule.PropertyMappingRule;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.util.ConfigHelper;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* @author jerrywang
|
||||
*
|
||||
*/
|
||||
@Slf4j
|
||||
public class MappingRuleFactory {
|
||||
private static final MappingRuleFactory instance = new MappingRuleFactory();
|
||||
private static final Pattern STR_PATTERN = Pattern.compile("\\$\\{(\\d+)}");
|
||||
private static final Pattern STR_NUM_PATTERN = Pattern.compile("#\\{(\\d+)}");
|
||||
private static final MappingRuleFactory instance = new MappingRuleFactory();
|
||||
|
||||
public static final MappingRuleFactory getInstance() {
|
||||
return instance;
|
||||
}
|
||||
|
||||
public static MappingRuleFactory getInstance() {
|
||||
return instance;
|
||||
@Deprecated
|
||||
public MappingRule create(Configuration config, ImportType type) {
|
||||
MappingRule rule = new MappingRule();
|
||||
rule.setId(config.getString(Key.ID));
|
||||
rule.setLabel(config.getString(Key.LABEL));
|
||||
if (type == ImportType.EDGE) {
|
||||
rule.setFrom(config.getString(Key.FROM));
|
||||
rule.setTo(config.getString(Key.TO));
|
||||
}
|
||||
|
||||
rule.setImportType(type);
|
||||
|
||||
List<Configuration> configurations = config.getListConfiguration(Key.PROPERTIES);
|
||||
if (configurations != null) {
|
||||
for (Configuration prop : config.getListConfiguration(Key.PROPERTIES)) {
|
||||
PropertyMappingRule propRule = new PropertyMappingRule();
|
||||
propRule.setKey(prop.getString(Key.PROP_KEY));
|
||||
propRule.setValue(prop.getString(Key.PROP_VALUE));
|
||||
propRule.setValueType(ValueType.fromShortName(prop.getString(Key.PROP_TYPE).toLowerCase()));
|
||||
rule.getProperties().add(propRule);
|
||||
}
|
||||
}
|
||||
|
||||
String propertiesJsonStr = config.getString(Key.PROPERTIES_JSON_STR, null);
|
||||
if (propertiesJsonStr != null) {
|
||||
rule.setPropertiesJsonStr(propertiesJsonStr);
|
||||
}
|
||||
|
||||
return rule;
|
||||
}
|
||||
|
||||
public MappingRule createV2(Configuration config) {
|
||||
try {
|
||||
ImportType type = ImportType.valueOf(config.getString(Key.IMPORT_TYPE));
|
||||
return createV2(config, type);
|
||||
} catch (NullPointerException e) {
|
||||
throw DataXException.asDataXException(GdbWriterErrorCode.CONFIG_ITEM_MISS, Key.IMPORT_TYPE);
|
||||
} catch (IllegalArgumentException e) {
|
||||
throw DataXException.asDataXException(GdbWriterErrorCode.BAD_CONFIG_VALUE, Key.IMPORT_TYPE);
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean isPattern(final String value, final MappingRule rule, final boolean checked) {
|
||||
if (checked) {
|
||||
return true;
|
||||
}
|
||||
public MappingRule createV2(Configuration config, ImportType type) {
|
||||
MappingRule rule = new MappingRule();
|
||||
|
||||
if (value == null || value.isEmpty()) {
|
||||
return false;
|
||||
}
|
||||
ConfigHelper.assertHasContent(config, Key.LABEL);
|
||||
rule.setLabel(config.getString(Key.LABEL));
|
||||
rule.setImportType(type);
|
||||
|
||||
Matcher m = STR_PATTERN.matcher(value);
|
||||
if (m.find()) {
|
||||
rule.setNumPattern(false);
|
||||
return true;
|
||||
}
|
||||
|
||||
m = STR_NUM_PATTERN.matcher(value);
|
||||
if (m.find()) {
|
||||
rule.setNumPattern(true);
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
@Deprecated
|
||||
public MappingRule create(final Configuration config, final ImportType type) {
|
||||
final MappingRule rule = new MappingRule();
|
||||
rule.setId(config.getString(Key.ID));
|
||||
rule.setLabel(config.getString(Key.LABEL));
|
||||
if (type == ImportType.EDGE) {
|
||||
rule.setFrom(config.getString(Key.FROM));
|
||||
rule.setTo(config.getString(Key.TO));
|
||||
}
|
||||
|
||||
rule.setImportType(type);
|
||||
|
||||
final List<Configuration> configurations = config.getListConfiguration(Key.PROPERTIES);
|
||||
if (configurations != null) {
|
||||
for (final Configuration prop : config.getListConfiguration(Key.PROPERTIES)) {
|
||||
final PropertyMappingRule propRule = new PropertyMappingRule();
|
||||
propRule.setKey(prop.getString(Key.PROP_KEY));
|
||||
propRule.setValue(prop.getString(Key.PROP_VALUE));
|
||||
propRule.setValueType(ValueType.fromShortName(prop.getString(Key.PROP_TYPE).toLowerCase()));
|
||||
rule.getProperties().add(propRule);
|
||||
}
|
||||
}
|
||||
|
||||
final String propertiesJsonStr = config.getString(Key.PROPERTIES_JSON_STR, null);
|
||||
if (propertiesJsonStr != null) {
|
||||
rule.setPropertiesJsonStr(propertiesJsonStr);
|
||||
}
|
||||
|
||||
return rule;
|
||||
}
|
||||
|
||||
public MappingRule createV2(final Configuration config) {
|
||||
try {
|
||||
final ImportType type = ImportType.valueOf(config.getString(Key.IMPORT_TYPE));
|
||||
return createV2(config, type);
|
||||
} catch (final NullPointerException e) {
|
||||
throw DataXException.asDataXException(GdbWriterErrorCode.CONFIG_ITEM_MISS, Key.IMPORT_TYPE);
|
||||
} catch (final IllegalArgumentException e) {
|
||||
throw DataXException.asDataXException(GdbWriterErrorCode.BAD_CONFIG_VALUE, Key.IMPORT_TYPE);
|
||||
}
|
||||
}
|
||||
|
||||
public MappingRule createV2(final Configuration config, final ImportType type) {
|
||||
final MappingRule rule = new MappingRule();
|
||||
boolean patternChecked = false;
|
||||
|
||||
ConfigHelper.assertHasContent(config, Key.LABEL);
|
||||
rule.setLabel(config.getString(Key.LABEL));
|
||||
rule.setImportType(type);
|
||||
patternChecked = isPattern(rule.getLabel(), rule, patternChecked);
|
||||
|
||||
IdTransRule srcTransRule = IdTransRule.none;
|
||||
IdTransRule srcTransRule = IdTransRule.none;
|
||||
IdTransRule dstTransRule = IdTransRule.none;
|
||||
if (type == ImportType.EDGE) {
|
||||
ConfigHelper.assertHasContent(config, Key.SRC_ID_TRANS_RULE);
|
||||
ConfigHelper.assertHasContent(config, Key.DST_ID_TRANS_RULE);
|
||||
ConfigHelper.assertHasContent(config, Key.SRC_ID_TRANS_RULE);
|
||||
ConfigHelper.assertHasContent(config, Key.DST_ID_TRANS_RULE);
|
||||
|
||||
srcTransRule = IdTransRule.valueOf(config.getString(Key.SRC_ID_TRANS_RULE));
|
||||
dstTransRule = IdTransRule.valueOf(config.getString(Key.DST_ID_TRANS_RULE));
|
||||
@@ -125,96 +94,88 @@ public class MappingRuleFactory {
|
||||
ConfigHelper.assertHasContent(config, Key.DST_LABEL);
|
||||
}
|
||||
}
|
||||
ConfigHelper.assertHasContent(config, Key.ID_TRANS_RULE);
|
||||
final IdTransRule transRule = IdTransRule.valueOf(config.getString(Key.ID_TRANS_RULE));
|
||||
ConfigHelper.assertHasContent(config, Key.ID_TRANS_RULE);
|
||||
IdTransRule transRule = IdTransRule.valueOf(config.getString(Key.ID_TRANS_RULE));
|
||||
|
||||
final List<Configuration> configurationList = config.getListConfiguration(Key.COLUMN);
|
||||
ConfigHelper.assertConfig(Key.COLUMN, () -> (configurationList != null && !configurationList.isEmpty()));
|
||||
for (final Configuration column : configurationList) {
|
||||
ConfigHelper.assertHasContent(column, Key.COLUMN_NAME);
|
||||
ConfigHelper.assertHasContent(column, Key.COLUMN_VALUE);
|
||||
ConfigHelper.assertHasContent(column, Key.COLUMN_TYPE);
|
||||
ConfigHelper.assertHasContent(column, Key.COLUMN_NODE_TYPE);
|
||||
List<Configuration> configurationList = config.getListConfiguration(Key.COLUMN);
|
||||
ConfigHelper.assertConfig(Key.COLUMN, () -> (configurationList != null && !configurationList.isEmpty()));
|
||||
for (Configuration column : configurationList) {
|
||||
ConfigHelper.assertHasContent(column, Key.COLUMN_NAME);
|
||||
ConfigHelper.assertHasContent(column, Key.COLUMN_VALUE);
|
||||
ConfigHelper.assertHasContent(column, Key.COLUMN_TYPE);
|
||||
ConfigHelper.assertHasContent(column, Key.COLUMN_NODE_TYPE);
|
||||
|
||||
final String columnValue = column.getString(Key.COLUMN_VALUE);
|
||||
final ColumnType columnType = ColumnType.valueOf(column.getString(Key.COLUMN_NODE_TYPE));
|
||||
if (columnValue == null || columnValue.isEmpty()) {
|
||||
// only allow edge empty id
|
||||
ConfigHelper.assertConfig("empty column value",
|
||||
() -> (type == ImportType.EDGE && columnType == ColumnType.primaryKey));
|
||||
}
|
||||
patternChecked = isPattern(columnValue, rule, patternChecked);
|
||||
String columnValue = column.getString(Key.COLUMN_VALUE);
|
||||
ColumnType columnType = ColumnType.valueOf(column.getString(Key.COLUMN_NODE_TYPE));
|
||||
if (columnValue == null || columnValue.isEmpty()) {
|
||||
// only allow edge empty id
|
||||
ConfigHelper.assertConfig("empty column value",
|
||||
() -> (type == ImportType.EDGE && columnType == ColumnType.primaryKey));
|
||||
}
|
||||
|
||||
if (columnType == ColumnType.primaryKey) {
|
||||
final ValueType propType = ValueType.fromShortName(column.getString(Key.COLUMN_TYPE));
|
||||
ConfigHelper.assertConfig("only string is allowed in primary key",
|
||||
() -> (propType == ValueType.STRING));
|
||||
if (columnType == ColumnType.primaryKey) {
|
||||
ValueType propType = ValueType.fromShortName(column.getString(Key.COLUMN_TYPE));
|
||||
ConfigHelper.assertConfig("only string is allowed in primary key", () -> (propType == ValueType.STRING));
|
||||
|
||||
if (transRule == IdTransRule.labelPrefix) {
|
||||
rule.setId(config.getString(Key.LABEL) + columnValue);
|
||||
} else {
|
||||
rule.setId(columnValue);
|
||||
}
|
||||
} else if (columnType == ColumnType.edgeJsonProperty || columnType == ColumnType.vertexJsonProperty) {
|
||||
// only support one json property in column
|
||||
ConfigHelper.assertConfig("multi JsonProperty", () -> (rule.getPropertiesJsonStr() == null));
|
||||
if (transRule == IdTransRule.labelPrefix) {
|
||||
rule.setId(config.getString(Key.LABEL) + columnValue);
|
||||
} else {
|
||||
rule.setId(columnValue);
|
||||
}
|
||||
} else if (columnType == ColumnType.edgeJsonProperty || columnType == ColumnType.vertexJsonProperty) {
|
||||
// only support one json property in column
|
||||
ConfigHelper.assertConfig("multi JsonProperty", () -> (rule.getPropertiesJsonStr() == null));
|
||||
|
||||
rule.setPropertiesJsonStr(columnValue);
|
||||
} else if (columnType == ColumnType.vertexProperty || columnType == ColumnType.edgeProperty
|
||||
|| columnType == ColumnType.vertexSetProperty) {
|
||||
final PropertyMappingRule propertyMappingRule = new PropertyMappingRule();
|
||||
rule.setPropertiesJsonStr(columnValue);
|
||||
} else if (columnType == ColumnType.vertexProperty || columnType == ColumnType.edgeProperty) {
|
||||
PropertyMappingRule propertyMappingRule = new PropertyMappingRule();
|
||||
|
||||
propertyMappingRule.setKey(column.getString(Key.COLUMN_NAME));
|
||||
propertyMappingRule.setValue(columnValue);
|
||||
final ValueType propType = ValueType.fromShortName(column.getString(Key.COLUMN_TYPE));
|
||||
ConfigHelper.assertConfig("unsupported property type", () -> propType != null);
|
||||
propertyMappingRule.setKey(column.getString(Key.COLUMN_NAME));
|
||||
propertyMappingRule.setValue(columnValue);
|
||||
ValueType propType = ValueType.fromShortName(column.getString(Key.COLUMN_TYPE));
|
||||
ConfigHelper.assertConfig("unsupported property type", () -> propType != null);
|
||||
|
||||
if (columnType == ColumnType.vertexSetProperty) {
|
||||
propertyMappingRule.setPType(Key.PropertyType.set);
|
||||
}
|
||||
propertyMappingRule.setValueType(propType);
|
||||
rule.getProperties().add(propertyMappingRule);
|
||||
} else if (columnType == ColumnType.srcPrimaryKey) {
|
||||
if (type != ImportType.EDGE) {
|
||||
continue;
|
||||
propertyMappingRule.setValueType(propType);
|
||||
rule.getProperties().add(propertyMappingRule);
|
||||
} else if (columnType == ColumnType.srcPrimaryKey) {
|
||||
if (type != ImportType.EDGE) {
|
||||
continue;
|
||||
}
|
||||
|
||||
final ValueType propType = ValueType.fromShortName(column.getString(Key.COLUMN_TYPE));
|
||||
ConfigHelper.assertConfig("only string is allowed in primary key",
|
||||
() -> (propType == ValueType.STRING));
|
||||
ValueType propType = ValueType.fromShortName(column.getString(Key.COLUMN_TYPE));
|
||||
ConfigHelper.assertConfig("only string is allowed in primary key", () -> (propType == ValueType.STRING));
|
||||
|
||||
if (srcTransRule == IdTransRule.labelPrefix) {
|
||||
rule.setFrom(config.getString(Key.SRC_LABEL) + columnValue);
|
||||
if (srcTransRule == IdTransRule.labelPrefix) {
|
||||
rule.setFrom(config.getString(Key.SRC_LABEL) + columnValue);
|
||||
} else {
|
||||
rule.setFrom(columnValue);
|
||||
}
|
||||
} else if (columnType == ColumnType.dstPrimaryKey) {
|
||||
} else if (columnType == ColumnType.dstPrimaryKey) {
|
||||
if (type != ImportType.EDGE) {
|
||||
continue;
|
||||
}
|
||||
|
||||
final ValueType propType = ValueType.fromShortName(column.getString(Key.COLUMN_TYPE));
|
||||
ConfigHelper.assertConfig("only string is allowed in primary key",
|
||||
() -> (propType == ValueType.STRING));
|
||||
ValueType propType = ValueType.fromShortName(column.getString(Key.COLUMN_TYPE));
|
||||
ConfigHelper.assertConfig("only string is allowed in primary key", () -> (propType == ValueType.STRING));
|
||||
|
||||
if (dstTransRule == IdTransRule.labelPrefix) {
|
||||
rule.setTo(config.getString(Key.DST_LABEL) + columnValue);
|
||||
} else {
|
||||
rule.setTo(columnValue);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (rule.getImportType() == ImportType.EDGE) {
|
||||
if (rule.getId() == null) {
|
||||
rule.setId("");
|
||||
log.info("edge id is missed, uuid be default");
|
||||
}
|
||||
ConfigHelper.assertConfig("to needed in edge", () -> (rule.getTo() != null));
|
||||
ConfigHelper.assertConfig("from needed in edge", () -> (rule.getFrom() != null));
|
||||
}
|
||||
ConfigHelper.assertConfig("id needed", () -> (rule.getId() != null));
|
||||
if (rule.getImportType() == ImportType.EDGE) {
|
||||
if (rule.getId() == null) {
|
||||
rule.setId("");
|
||||
log.info("edge id is missed, uuid be default");
|
||||
}
|
||||
ConfigHelper.assertConfig("to needed in edge", () -> (rule.getTo() != null));
|
||||
ConfigHelper.assertConfig("from needed in edge", () -> (rule.getFrom() != null));
|
||||
}
|
||||
ConfigHelper.assertConfig("id needed", () -> (rule.getId() != null));
|
||||
|
||||
return rule;
|
||||
}
|
||||
return rule;
|
||||
}
|
||||
}
|
||||
|
||||
+48
-54
@@ -8,7 +8,6 @@ import java.util.Map;
|
||||
import java.util.function.Function;
|
||||
|
||||
import com.alibaba.datax.common.element.Column;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
/**
|
||||
@@ -17,61 +16,56 @@ import lombok.extern.slf4j.Slf4j;
|
||||
*/
|
||||
@Slf4j
|
||||
public enum ValueType {
|
||||
/**
|
||||
* property value type
|
||||
*/
|
||||
INT(Integer.class, "int", Column::asLong, Integer::valueOf),
|
||||
INTEGER(Integer.class, "integer", Column::asLong, Integer::valueOf),
|
||||
LONG(Long.class, "long", Column::asLong, Long::valueOf),
|
||||
DOUBLE(Double.class, "double", Column::asDouble, Double::valueOf),
|
||||
FLOAT(Float.class, "float", Column::asDouble, Float::valueOf),
|
||||
BOOLEAN(Boolean.class, "boolean", Column::asBoolean, Boolean::valueOf),
|
||||
STRING(String.class, "string", Column::asString, String::valueOf);
|
||||
INT(Integer.class, "int", Column::asLong, Integer::valueOf),
|
||||
LONG(Long.class, "long", Column::asLong, Long::valueOf),
|
||||
DOUBLE(Double.class, "double", Column::asDouble, Double::valueOf),
|
||||
FLOAT(Float.class, "float", Column::asDouble, Float::valueOf),
|
||||
BOOLEAN(Boolean.class, "boolean", Column::asBoolean, Boolean::valueOf),
|
||||
STRING(String.class, "string", Column::asString, String::valueOf);
|
||||
|
||||
private Class<?> type = null;
|
||||
private String shortName = null;
|
||||
private Function<Column, Object> columnFunc = null;
|
||||
private Function<String, Object> fromStrFunc = null;
|
||||
private Class<?> type = null;
|
||||
private String shortName = null;
|
||||
private Function<Column, Object> columnFunc = null;
|
||||
private Function<String, Object> fromStrFunc = null;
|
||||
|
||||
private ValueType(final Class<?> type, final String name, final Function<Column, Object> columnFunc,
|
||||
final Function<String, Object> fromStrFunc) {
|
||||
this.type = type;
|
||||
this.shortName = name;
|
||||
this.columnFunc = columnFunc;
|
||||
this.fromStrFunc = fromStrFunc;
|
||||
private ValueType(Class<?> type, String name, Function<Column, Object> columnFunc, Function<String, Object> fromStrFunc) {
|
||||
this.type = type;
|
||||
this.shortName = name;
|
||||
this.columnFunc = columnFunc;
|
||||
this.fromStrFunc = fromStrFunc;
|
||||
|
||||
ValueTypeHolder.shortName2type.put(name, this);
|
||||
}
|
||||
|
||||
public static ValueType fromShortName(String name) {
|
||||
return ValueTypeHolder.shortName2type.get(name);
|
||||
}
|
||||
|
||||
ValueTypeHolder.shortName2type.put(name, this);
|
||||
}
|
||||
public Class<?> type() {
|
||||
return this.type;
|
||||
}
|
||||
|
||||
public String shortName() {
|
||||
return this.shortName;
|
||||
}
|
||||
|
||||
public Object applyColumn(Column column) {
|
||||
try {
|
||||
if (column == null) {
|
||||
return null;
|
||||
}
|
||||
return columnFunc.apply(column);
|
||||
} catch (Exception e) {
|
||||
log.error("applyColumn error {}, column {}", e.toString(), column);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
public Object fromStrFunc(String str) {
|
||||
return fromStrFunc.apply(str);
|
||||
}
|
||||
|
||||
public static ValueType fromShortName(final String name) {
|
||||
return ValueTypeHolder.shortName2type.get(name);
|
||||
}
|
||||
|
||||
public Class<?> type() {
|
||||
return this.type;
|
||||
}
|
||||
|
||||
public String shortName() {
|
||||
return this.shortName;
|
||||
}
|
||||
|
||||
public Object applyColumn(final Column column) {
|
||||
try {
|
||||
if (column == null) {
|
||||
return null;
|
||||
}
|
||||
return this.columnFunc.apply(column);
|
||||
} catch (final Exception e) {
|
||||
log.error("applyColumn error {}, column {}", e.toString(), column);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
public Object fromStrFunc(final String str) {
|
||||
return this.fromStrFunc.apply(str);
|
||||
}
|
||||
|
||||
private static class ValueTypeHolder {
|
||||
private static Map<String, ValueType> shortName2type = new HashMap<>();
|
||||
}
|
||||
private static class ValueTypeHolder {
|
||||
private static Map<String, ValueType> shortName2type = new HashMap<>();
|
||||
}
|
||||
}
|
||||
|
||||
+114
-114
@@ -3,24 +3,20 @@
|
||||
*/
|
||||
package com.alibaba.datax.plugin.writer.gdbwriter.model;
|
||||
|
||||
import static com.alibaba.datax.plugin.writer.gdbwriter.client.GdbWriterConfig.DEFAULT_BATCH_PROPERTY_NUM;
|
||||
import static com.alibaba.datax.plugin.writer.gdbwriter.client.GdbWriterConfig.MAX_REQUEST_LENGTH;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.apache.tinkerpop.gremlin.driver.Client;
|
||||
import org.apache.tinkerpop.gremlin.driver.Cluster;
|
||||
import org.apache.tinkerpop.gremlin.driver.RequestOptions;
|
||||
import org.apache.tinkerpop.gremlin.driver.ResultSet;
|
||||
import org.apache.tinkerpop.gremlin.driver.ser.Serializers;
|
||||
|
||||
import com.alibaba.datax.common.util.Configuration;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.Key;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.client.GdbWriterConfig;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.tinkerpop.gremlin.driver.Client;
|
||||
import org.apache.tinkerpop.gremlin.driver.Cluster;
|
||||
import org.apache.tinkerpop.gremlin.driver.RequestOptions;
|
||||
import org.apache.tinkerpop.gremlin.driver.ResultSet;
|
||||
import org.apache.tinkerpop.gremlin.driver.ser.Serializers;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* @author jerrywang
|
||||
@@ -28,124 +24,128 @@ import lombok.extern.slf4j.Slf4j;
|
||||
*/
|
||||
@Slf4j
|
||||
public abstract class AbstractGdbGraph implements GdbGraph {
|
||||
private final static int DEFAULT_TIMEOUT = 30000;
|
||||
private final static int DEFAULT_TIMEOUT = 30000;
|
||||
|
||||
protected Client client = null;
|
||||
protected Key.UpdateMode updateMode = Key.UpdateMode.INSERT;
|
||||
protected int propertiesBatchNum = DEFAULT_BATCH_PROPERTY_NUM;
|
||||
protected boolean session = false;
|
||||
protected int maxRequestLength = GdbWriterConfig.MAX_REQUEST_LENGTH;
|
||||
protected Client client = null;
|
||||
protected Key.UpdateMode updateMode = Key.UpdateMode.INSERT;
|
||||
protected int propertiesBatchNum = GdbWriterConfig.DEFAULT_BATCH_PROPERTY_NUM;
|
||||
protected boolean session = false;
|
||||
|
||||
protected AbstractGdbGraph() {}
|
||||
|
||||
protected AbstractGdbGraph(final Configuration config, final boolean session) {
|
||||
initClient(config, session);
|
||||
}
|
||||
protected AbstractGdbGraph() {}
|
||||
|
||||
protected void initClient(final Configuration config, final boolean session) {
|
||||
this.updateMode = Key.UpdateMode.valueOf(config.getString(Key.UPDATE_MODE, "INSERT"));
|
||||
log.info("init graphdb client");
|
||||
final String host = config.getString(Key.HOST);
|
||||
final int port = config.getInt(Key.PORT);
|
||||
final String username = config.getString(Key.USERNAME);
|
||||
final String password = config.getString(Key.PASSWORD);
|
||||
int maxDepthPerConnection =
|
||||
config.getInt(Key.MAX_IN_PROCESS_PER_CONNECTION, GdbWriterConfig.DEFAULT_MAX_IN_PROCESS_PER_CONNECTION);
|
||||
protected AbstractGdbGraph(Configuration config, boolean session) {
|
||||
initClient(config, session);
|
||||
}
|
||||
|
||||
int maxConnectionPoolSize =
|
||||
config.getInt(Key.MAX_CONNECTION_POOL_SIZE, GdbWriterConfig.DEFAULT_MAX_CONNECTION_POOL_SIZE);
|
||||
protected void initClient(Configuration config, boolean session) {
|
||||
updateMode = Key.UpdateMode.valueOf(config.getString(Key.UPDATE_MODE, "INSERT"));
|
||||
log.info("init graphdb client");
|
||||
String host = config.getString(Key.HOST);
|
||||
int port = config.getInt(Key.PORT);
|
||||
String username = config.getString(Key.USERNAME);
|
||||
String password = config.getString(Key.PASSWORD);
|
||||
int maxDepthPerConnection = config.getInt(Key.MAX_IN_PROCESS_PER_CONNECTION,
|
||||
GdbWriterConfig.DEFAULT_MAX_IN_PROCESS_PER_CONNECTION);
|
||||
|
||||
int maxSimultaneousUsagePerConnection = config.getInt(Key.MAX_SIMULTANEOUS_USAGE_PER_CONNECTION,
|
||||
GdbWriterConfig.DEFAULT_MAX_SIMULTANEOUS_USAGE_PER_CONNECTION);
|
||||
int maxConnectionPoolSize = config.getInt(Key.MAX_CONNECTION_POOL_SIZE,
|
||||
GdbWriterConfig.DEFAULT_MAX_CONNECTION_POOL_SIZE);
|
||||
|
||||
this.session = session;
|
||||
if (this.session) {
|
||||
maxConnectionPoolSize = GdbWriterConfig.DEFAULT_MAX_CONNECTION_POOL_SIZE;
|
||||
maxDepthPerConnection = GdbWriterConfig.DEFAULT_MAX_IN_PROCESS_PER_CONNECTION;
|
||||
maxSimultaneousUsagePerConnection = GdbWriterConfig.DEFAULT_MAX_SIMULTANEOUS_USAGE_PER_CONNECTION;
|
||||
}
|
||||
int maxSimultaneousUsagePerConnection = config.getInt(Key.MAX_SIMULTANEOUS_USAGE_PER_CONNECTION,
|
||||
GdbWriterConfig.DEFAULT_MAX_SIMULTANEOUS_USAGE_PER_CONNECTION);
|
||||
|
||||
try {
|
||||
final Cluster cluster = Cluster.build(host).port(port).credentials(username, password)
|
||||
.serializer(Serializers.GRAPHBINARY_V1D0).maxContentLength(1048576)
|
||||
.maxInProcessPerConnection(maxDepthPerConnection).minInProcessPerConnection(0)
|
||||
.maxConnectionPoolSize(maxConnectionPoolSize).minConnectionPoolSize(maxConnectionPoolSize)
|
||||
.maxSimultaneousUsagePerConnection(maxSimultaneousUsagePerConnection).resultIterationBatchSize(64)
|
||||
.create();
|
||||
this.client = session ? cluster.connect(UUID.randomUUID().toString()).init() : cluster.connect().init();
|
||||
warmClient(maxConnectionPoolSize * maxDepthPerConnection);
|
||||
} catch (final RuntimeException e) {
|
||||
log.error("Failed to connect to GDB {}:{}, due to {}", host, port, e);
|
||||
throw e;
|
||||
}
|
||||
this.session = session;
|
||||
if (this.session) {
|
||||
maxConnectionPoolSize = GdbWriterConfig.DEFAULT_MAX_CONNECTION_POOL_SIZE;
|
||||
maxDepthPerConnection = GdbWriterConfig.DEFAULT_MAX_IN_PROCESS_PER_CONNECTION;
|
||||
maxSimultaneousUsagePerConnection = GdbWriterConfig.DEFAULT_MAX_SIMULTANEOUS_USAGE_PER_CONNECTION;
|
||||
}
|
||||
|
||||
this.propertiesBatchNum = config.getInt(Key.MAX_PROPERTIES_BATCH_NUM, DEFAULT_BATCH_PROPERTY_NUM);
|
||||
this.maxRequestLength = config.getInt(Key.MAX_GDB_REQUEST_LENGTH, MAX_REQUEST_LENGTH);
|
||||
}
|
||||
try {
|
||||
Cluster cluster = Cluster.build(host).port(port).credentials(username, password)
|
||||
.serializer(Serializers.GRAPHBINARY_V1D0)
|
||||
.maxContentLength(1048576)
|
||||
.maxInProcessPerConnection(maxDepthPerConnection)
|
||||
.minInProcessPerConnection(0)
|
||||
.maxConnectionPoolSize(maxConnectionPoolSize)
|
||||
.minConnectionPoolSize(maxConnectionPoolSize)
|
||||
.maxSimultaneousUsagePerConnection(maxSimultaneousUsagePerConnection)
|
||||
.resultIterationBatchSize(64)
|
||||
.create();
|
||||
client = session ? cluster.connect(UUID.randomUUID().toString()).init() : cluster.connect().init();
|
||||
warmClient(maxConnectionPoolSize*maxDepthPerConnection);
|
||||
} catch (RuntimeException e) {
|
||||
log.error("Failed to connect to GDB {}:{}, due to {}", host, port, e);
|
||||
throw e;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param dsl
|
||||
* @param parameters
|
||||
*/
|
||||
protected void runInternal(final String dsl, final Map<String, Object> parameters) throws Exception {
|
||||
final RequestOptions.Builder options = RequestOptions.build().timeout(DEFAULT_TIMEOUT);
|
||||
if (parameters != null && !parameters.isEmpty()) {
|
||||
parameters.forEach(options::addParameter);
|
||||
}
|
||||
propertiesBatchNum = config.getInt(Key.MAX_PROPERTIES_BATCH_NUM, GdbWriterConfig.DEFAULT_BATCH_PROPERTY_NUM);
|
||||
}
|
||||
|
||||
final ResultSet results = this.client.submitAsync(dsl, options.create()).get(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
|
||||
results.all().get(DEFAULT_TIMEOUT + 1000, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
void beginTx() {
|
||||
if (!this.session) {
|
||||
return;
|
||||
}
|
||||
/**
|
||||
* @param dsl
|
||||
* @param parameters
|
||||
*/
|
||||
protected void runInternal(String dsl, final Map<String, Object> parameters) throws Exception {
|
||||
RequestOptions.Builder options = RequestOptions.build().timeout(DEFAULT_TIMEOUT);
|
||||
if (parameters != null && !parameters.isEmpty()) {
|
||||
parameters.forEach(options::addParameter);
|
||||
}
|
||||
|
||||
final String dsl = "g.tx().open()";
|
||||
this.client.submit(dsl).all().join();
|
||||
}
|
||||
ResultSet results = client.submitAsync(dsl, options.create()).get(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
|
||||
results.all().get(DEFAULT_TIMEOUT + 1000, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
void doCommit() {
|
||||
if (!this.session) {
|
||||
return;
|
||||
}
|
||||
void beginTx() {
|
||||
if (!session) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
final String dsl = "g.tx().commit()";
|
||||
this.client.submit(dsl).all().join();
|
||||
} catch (final Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
String dsl = "g.tx().open()";
|
||||
client.submit(dsl).all().join();
|
||||
}
|
||||
|
||||
void doRollback() {
|
||||
if (!this.session) {
|
||||
return;
|
||||
}
|
||||
void doCommit() {
|
||||
if (!session) {
|
||||
return;
|
||||
}
|
||||
|
||||
final String dsl = "g.tx().rollback()";
|
||||
this.client.submit(dsl).all().join();
|
||||
}
|
||||
try {
|
||||
String dsl = "g.tx().commit()";
|
||||
client.submit(dsl).all().join();
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
private void warmClient(final int num) {
|
||||
try {
|
||||
beginTx();
|
||||
runInternal("g.V('test')", null);
|
||||
doCommit();
|
||||
log.info("warm graphdb client over");
|
||||
} catch (final Exception e) {
|
||||
log.error("warmClient error");
|
||||
doRollback();
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
void doRollback() {
|
||||
if (!session) {
|
||||
return;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
if (this.client != null) {
|
||||
log.info("close graphdb client");
|
||||
this.client.close();
|
||||
}
|
||||
}
|
||||
String dsl = "g.tx().rollback()";
|
||||
client.submit(dsl).all().join();
|
||||
}
|
||||
|
||||
private void warmClient(int num) {
|
||||
try {
|
||||
beginTx();
|
||||
runInternal("g.V('test')", null);
|
||||
doCommit();
|
||||
log.info("warm graphdb client over");
|
||||
} catch (Exception e) {
|
||||
log.error("warmClient error");
|
||||
doRollback();
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
if (client != null) {
|
||||
log.info("close graphdb client");
|
||||
client.close();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+4
-28
@@ -3,8 +3,7 @@
|
||||
*/
|
||||
package com.alibaba.datax.plugin.writer.gdbwriter.model;
|
||||
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.mapping.MapperConfig;
|
||||
|
||||
import lombok.Data;
|
||||
import lombok.EqualsAndHashCode;
|
||||
import lombok.ToString;
|
||||
|
||||
@@ -12,33 +11,10 @@ import lombok.ToString;
|
||||
* @author jerrywang
|
||||
*
|
||||
*/
|
||||
@Data
|
||||
@EqualsAndHashCode(callSuper = true)
|
||||
@ToString(callSuper = true)
|
||||
public class GdbEdge extends GdbElement {
|
||||
private String from = null;
|
||||
private String to = null;
|
||||
|
||||
public String getFrom() {
|
||||
return this.from;
|
||||
}
|
||||
|
||||
public void setFrom(final String from) {
|
||||
final int maxIdLength = MapperConfig.getInstance().getMaxIdLength();
|
||||
if (from.length() > maxIdLength) {
|
||||
throw new IllegalArgumentException("from length over limit(" + maxIdLength + ")");
|
||||
}
|
||||
this.from = from;
|
||||
}
|
||||
|
||||
public String getTo() {
|
||||
return this.to;
|
||||
}
|
||||
|
||||
public void setTo(final String to) {
|
||||
final int maxIdLength = MapperConfig.getInstance().getMaxIdLength();
|
||||
if (to.length() > maxIdLength) {
|
||||
throw new IllegalArgumentException("to length over limit(" + maxIdLength + ")");
|
||||
}
|
||||
this.to = to;
|
||||
}
|
||||
private String from = null;
|
||||
private String to = null;
|
||||
}
|
||||
|
||||
+7
-96
@@ -3,107 +3,18 @@
|
||||
*/
|
||||
package com.alibaba.datax.plugin.writer.gdbwriter.model;
|
||||
|
||||
import java.util.LinkedList;
|
||||
import java.util.List;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.Key.PropertyType;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.mapping.MapperConfig;
|
||||
import lombok.Data;
|
||||
|
||||
/**
|
||||
* @author jerrywang
|
||||
*
|
||||
*/
|
||||
@Data
|
||||
public class GdbElement {
|
||||
private String id = null;
|
||||
private String label = null;
|
||||
private List<GdbProperty> properties = new LinkedList<>();
|
||||
|
||||
public String getId() {
|
||||
return this.id;
|
||||
}
|
||||
|
||||
public void setId(final String id) {
|
||||
final int maxIdLength = MapperConfig.getInstance().getMaxIdLength();
|
||||
if (id.length() > maxIdLength) {
|
||||
throw new IllegalArgumentException("id length over limit(" + maxIdLength + ")");
|
||||
}
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
public String getLabel() {
|
||||
return this.label;
|
||||
}
|
||||
|
||||
public void setLabel(final String label) {
|
||||
final int maxLabelLength = MapperConfig.getInstance().getMaxLabelLength();
|
||||
if (label.length() > maxLabelLength) {
|
||||
throw new IllegalArgumentException("label length over limit(" + maxLabelLength + ")");
|
||||
}
|
||||
this.label = label;
|
||||
}
|
||||
|
||||
public List<GdbProperty> getProperties() {
|
||||
return this.properties;
|
||||
}
|
||||
|
||||
public void addProperty(final String propKey, final Object propValue, final PropertyType card) {
|
||||
if (propKey == null || propValue == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
final int maxPropKeyLength = MapperConfig.getInstance().getMaxPropKeyLength();
|
||||
if (propKey.length() > maxPropKeyLength) {
|
||||
throw new IllegalArgumentException("property key length over limit(" + maxPropKeyLength + ")");
|
||||
}
|
||||
if (propValue instanceof String) {
|
||||
final int maxPropValueLength = MapperConfig.getInstance().getMaxPropValueLength();
|
||||
if (((String)propValue).length() > maxPropKeyLength) {
|
||||
throw new IllegalArgumentException("property value length over limit(" + maxPropValueLength + ")");
|
||||
}
|
||||
}
|
||||
|
||||
this.properties.add(new GdbProperty(propKey, propValue, card));
|
||||
}
|
||||
|
||||
public void addProperty(final String propKey, final Object propValue) {
|
||||
addProperty(propKey, propValue, PropertyType.single);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
final StringBuffer sb = new StringBuffer(this.id + "[" + this.label + "]{");
|
||||
this.properties.forEach(n -> {
|
||||
sb.append(n.cardinality.name());
|
||||
sb.append("[");
|
||||
sb.append(n.key);
|
||||
sb.append(" - ");
|
||||
sb.append(String.valueOf(n.value));
|
||||
sb.append("]");
|
||||
});
|
||||
return sb.toString();
|
||||
}
|
||||
|
||||
public static class GdbProperty {
|
||||
private String key;
|
||||
private Object value;
|
||||
private PropertyType cardinality;
|
||||
|
||||
private GdbProperty(final String key, final Object value, final PropertyType card) {
|
||||
this.key = key;
|
||||
this.value = value;
|
||||
this.cardinality = card;
|
||||
}
|
||||
|
||||
public PropertyType getCardinality() {
|
||||
return this.cardinality;
|
||||
}
|
||||
|
||||
public String getKey() {
|
||||
return this.key;
|
||||
}
|
||||
|
||||
public Object getValue() {
|
||||
return this.value;
|
||||
}
|
||||
}
|
||||
String id = null;
|
||||
String label = null;
|
||||
Map<String, Object> properties = new HashMap<>();
|
||||
}
|
||||
|
||||
+5
-6
@@ -3,19 +3,18 @@
|
||||
*/
|
||||
package com.alibaba.datax.plugin.writer.gdbwriter.model;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import com.alibaba.datax.common.element.Record;
|
||||
|
||||
import groovy.lang.Tuple2;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* @author jerrywang
|
||||
*
|
||||
*/
|
||||
public interface GdbGraph extends AutoCloseable {
|
||||
List<Tuple2<Record, Exception>> add(List<Tuple2<Record, GdbElement>> records);
|
||||
List<Tuple2<Record, Exception>> add(List<Tuple2<Record, GdbElement>> records);
|
||||
|
||||
@Override
|
||||
void close();
|
||||
@Override
|
||||
void close();
|
||||
}
|
||||
|
||||
+157
-181
@@ -3,17 +3,15 @@
|
||||
*/
|
||||
package com.alibaba.datax.plugin.writer.gdbwriter.model;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Random;
|
||||
import java.util.*;
|
||||
|
||||
import com.alibaba.datax.common.element.Record;
|
||||
import com.alibaba.datax.common.util.Configuration;
|
||||
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.Key;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.util.GdbDuplicateIdException;
|
||||
|
||||
import com.github.benmanes.caffeine.cache.Cache;
|
||||
import com.github.benmanes.caffeine.cache.Caffeine;
|
||||
import groovy.lang.Tuple2;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
@@ -23,198 +21,176 @@ import lombok.extern.slf4j.Slf4j;
|
||||
*/
|
||||
@Slf4j
|
||||
public class ScriptGdbGraph extends AbstractGdbGraph {
|
||||
private static final String VAR_PREFIX = "GDB___";
|
||||
private static final String VAR_ID = VAR_PREFIX + "id";
|
||||
private static final String VAR_LABEL = VAR_PREFIX + "label";
|
||||
private static final String VAR_FROM = VAR_PREFIX + "from";
|
||||
private static final String VAR_TO = VAR_PREFIX + "to";
|
||||
private static final String VAR_PROP_KEY = VAR_PREFIX + "PK";
|
||||
private static final String VAR_PROP_VALUE = VAR_PREFIX + "PV";
|
||||
private static final String ADD_V_START = "g.addV(" + VAR_LABEL + ").property(id, " + VAR_ID + ")";
|
||||
private static final String ADD_E_START =
|
||||
"g.addE(" + VAR_LABEL + ").property(id, " + VAR_ID + ").from(V(" + VAR_FROM + ")).to(V(" + VAR_TO + "))";
|
||||
private static final String VAR_PREFIX = "GDB___";
|
||||
private static final String VAR_ID = VAR_PREFIX + "id";
|
||||
private static final String VAR_LABEL = VAR_PREFIX + "label";
|
||||
private static final String VAR_FROM = VAR_PREFIX + "from";
|
||||
private static final String VAR_TO = VAR_PREFIX + "to";
|
||||
private static final String VAR_PROP_KEY = VAR_PREFIX + "PK";
|
||||
private static final String VAR_PROP_VALUE = VAR_PREFIX + "PV";
|
||||
private static final String ADD_V_START = "g.addV(" + VAR_LABEL + ").property(id, " + VAR_ID + ")";
|
||||
private static final String ADD_E_START = "g.addE(" + VAR_LABEL + ").property(id, " + VAR_ID + ").from(V("
|
||||
+ VAR_FROM + ")).to(V(" + VAR_TO + "))";
|
||||
|
||||
private static final String UPDATE_V_START = "g.V(" + VAR_ID + ")";
|
||||
private static final String UPDATE_E_START = "g.E(" + VAR_ID + ")";
|
||||
private static final String UPDATE_V_START = "g.V("+VAR_ID+")";
|
||||
private static final String UPDATE_E_START = "g.E("+VAR_ID+")";
|
||||
|
||||
private Random random;
|
||||
private Cache<Integer, String> propertyCache;
|
||||
private Random random;
|
||||
|
||||
public ScriptGdbGraph() {
|
||||
this.random = new Random();
|
||||
}
|
||||
public ScriptGdbGraph() {
|
||||
propertyCache = Caffeine.newBuilder().maximumSize(1024).build();
|
||||
random = new Random();
|
||||
}
|
||||
|
||||
public ScriptGdbGraph(final Configuration config, final boolean session) {
|
||||
super(config, session);
|
||||
public ScriptGdbGraph(Configuration config, boolean session) {
|
||||
super(config, session);
|
||||
|
||||
this.random = new Random();
|
||||
log.info("Init as ScriptGdbGraph.");
|
||||
}
|
||||
propertyCache = Caffeine.newBuilder().maximumSize(1024).build();
|
||||
random = new Random();
|
||||
|
||||
/**
|
||||
* Apply list of {@link GdbElement} to GDB, return the failed records
|
||||
*
|
||||
* @param records
|
||||
* list of element to apply
|
||||
* @return
|
||||
*/
|
||||
@Override
|
||||
public List<Tuple2<Record, Exception>> add(final List<Tuple2<Record, GdbElement>> records) {
|
||||
final List<Tuple2<Record, Exception>> errors = new ArrayList<>();
|
||||
try {
|
||||
beginTx();
|
||||
for (final Tuple2<Record, GdbElement> elementTuple2 : records) {
|
||||
try {
|
||||
addInternal(elementTuple2.getSecond());
|
||||
} catch (final Exception e) {
|
||||
errors.add(new Tuple2<>(elementTuple2.getFirst(), e));
|
||||
}
|
||||
}
|
||||
doCommit();
|
||||
} catch (final Exception ex) {
|
||||
doRollback();
|
||||
throw new RuntimeException(ex);
|
||||
}
|
||||
return errors;
|
||||
}
|
||||
log.info("Init as ScriptGdbGraph.");
|
||||
}
|
||||
|
||||
private void addInternal(final GdbElement element) {
|
||||
try {
|
||||
addInternal(element, false);
|
||||
} catch (final GdbDuplicateIdException e) {
|
||||
if (this.updateMode == Key.UpdateMode.SKIP) {
|
||||
log.debug("Skip duplicate id {}", element.getId());
|
||||
} else if (this.updateMode == Key.UpdateMode.INSERT) {
|
||||
throw new RuntimeException(e);
|
||||
} else if (this.updateMode == Key.UpdateMode.MERGE) {
|
||||
if (element.getProperties().isEmpty()) {
|
||||
return;
|
||||
}
|
||||
/**
|
||||
* Apply list of {@link GdbElement} to GDB, return the failed records
|
||||
* @param records list of element to apply
|
||||
* @return
|
||||
*/
|
||||
@Override
|
||||
public List<Tuple2<Record, Exception>> add(List<Tuple2<Record, GdbElement>> records) {
|
||||
List<Tuple2<Record, Exception>> errors = new ArrayList<>();
|
||||
try {
|
||||
beginTx();
|
||||
for (Tuple2<Record, GdbElement> elementTuple2 : records) {
|
||||
try {
|
||||
addInternal(elementTuple2.getSecond());
|
||||
} catch (Exception e) {
|
||||
errors.add(new Tuple2<>(elementTuple2.getFirst(), e));
|
||||
}
|
||||
}
|
||||
doCommit();
|
||||
} catch (Exception ex) {
|
||||
doRollback();
|
||||
throw new RuntimeException(ex);
|
||||
}
|
||||
return errors;
|
||||
}
|
||||
|
||||
try {
|
||||
addInternal(element, true);
|
||||
} catch (final GdbDuplicateIdException e1) {
|
||||
log.error("duplicate id {} while update...", element.getId());
|
||||
throw new RuntimeException(e1);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
private void addInternal(GdbElement element) {
|
||||
try {
|
||||
addInternal(element, false);
|
||||
} catch (GdbDuplicateIdException e) {
|
||||
if (updateMode == Key.UpdateMode.SKIP) {
|
||||
log.debug("Skip duplicate id {}", element.getId());
|
||||
} else if (updateMode == Key.UpdateMode.INSERT) {
|
||||
throw new RuntimeException(e);
|
||||
} else if (updateMode == Key.UpdateMode.MERGE) {
|
||||
if (element.getProperties().isEmpty()) {
|
||||
return;
|
||||
}
|
||||
|
||||
private void addInternal(final GdbElement element, final boolean update) throws GdbDuplicateIdException {
|
||||
boolean firstAdd = !update;
|
||||
final boolean isVertex = (element instanceof GdbVertex);
|
||||
final List<GdbElement.GdbProperty> params = element.getProperties();
|
||||
final List<GdbElement.GdbProperty> subParams = new ArrayList<>(this.propertiesBatchNum);
|
||||
try {
|
||||
addInternal(element, true);
|
||||
} catch (GdbDuplicateIdException e1) {
|
||||
log.error("duplicate id {} while update...", element.getId());
|
||||
throw new RuntimeException(e1);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
final int idLength = element.getId().length();
|
||||
int attachLength = element.getLabel().length();
|
||||
if (element instanceof GdbEdge) {
|
||||
attachLength += ((GdbEdge)element).getFrom().length();
|
||||
attachLength += ((GdbEdge)element).getTo().length();
|
||||
}
|
||||
private void addInternal(GdbElement element, boolean update) throws GdbDuplicateIdException {
|
||||
Map<String, Object> params = element.getProperties();
|
||||
Map<String, Object> subParams = new HashMap<>(propertiesBatchNum);
|
||||
boolean firstAdd = !update;
|
||||
boolean isVertex = (element instanceof GdbVertex);
|
||||
|
||||
int requestLength = idLength;
|
||||
for (final GdbElement.GdbProperty entry : params) {
|
||||
final String propKey = entry.getKey();
|
||||
final Object propValue = entry.getValue();
|
||||
for (Map.Entry<String, Object> entry : params.entrySet()) {
|
||||
subParams.put(entry.getKey(), entry.getValue());
|
||||
if (subParams.size() >= propertiesBatchNum) {
|
||||
setGraphDbElement(element, subParams, isVertex, firstAdd);
|
||||
firstAdd = false;
|
||||
subParams.clear();
|
||||
}
|
||||
}
|
||||
if (!subParams.isEmpty() || firstAdd) {
|
||||
setGraphDbElement(element, subParams, isVertex, firstAdd);
|
||||
}
|
||||
}
|
||||
|
||||
int appendLength = propKey.length();
|
||||
if (propValue instanceof String) {
|
||||
appendLength += ((String)propValue).length();
|
||||
}
|
||||
private Tuple2<String, Map<String, Object>> buildDsl(GdbElement element,
|
||||
Map<String, Object> properties,
|
||||
boolean isVertex, boolean firstAdd) {
|
||||
Map<String, Object> params = new HashMap<>();
|
||||
|
||||
if (checkSplitDsl(firstAdd, requestLength, attachLength, appendLength, subParams.size())) {
|
||||
setGraphDbElement(element, subParams, isVertex, firstAdd);
|
||||
firstAdd = false;
|
||||
subParams.clear();
|
||||
requestLength = idLength;
|
||||
}
|
||||
String dslPropertyPart = propertyCache.get(properties.size(), keys -> {
|
||||
final StringBuilder sb = new StringBuilder();
|
||||
for (int i = 0; i < keys; i++) {
|
||||
sb.append(".property(").append(VAR_PROP_KEY).append(i)
|
||||
.append(", ").append(VAR_PROP_VALUE).append(i).append(")");
|
||||
}
|
||||
return sb.toString();
|
||||
});
|
||||
|
||||
requestLength += appendLength;
|
||||
subParams.add(entry);
|
||||
}
|
||||
if (!subParams.isEmpty() || firstAdd) {
|
||||
checkSplitDsl(firstAdd, requestLength, attachLength, 0, 0);
|
||||
setGraphDbElement(element, subParams, isVertex, firstAdd);
|
||||
}
|
||||
}
|
||||
String dsl;
|
||||
if (isVertex) {
|
||||
dsl = (firstAdd ? ADD_V_START : UPDATE_V_START) + dslPropertyPart;
|
||||
} else {
|
||||
dsl = (firstAdd ? ADD_E_START : UPDATE_E_START) + dslPropertyPart;
|
||||
if (firstAdd) {
|
||||
params.put(VAR_FROM, ((GdbEdge)element).getFrom());
|
||||
params.put(VAR_TO, ((GdbEdge)element).getTo());
|
||||
}
|
||||
}
|
||||
|
||||
private boolean checkSplitDsl(final boolean firstAdd, final int requestLength, final int attachLength, final int appendLength,
|
||||
final int propNum) {
|
||||
final int length = firstAdd ? requestLength + attachLength : requestLength;
|
||||
if (length > this.maxRequestLength) {
|
||||
throw new IllegalArgumentException("request length over limit(" + this.maxRequestLength + ")");
|
||||
}
|
||||
return length + appendLength > this.maxRequestLength || propNum >= this.propertiesBatchNum;
|
||||
}
|
||||
int index = 0;
|
||||
for (Map.Entry<String, Object> entry : properties.entrySet()) {
|
||||
params.put(VAR_PROP_KEY+index, entry.getKey());
|
||||
params.put(VAR_PROP_VALUE+index, entry.getValue());
|
||||
index++;
|
||||
}
|
||||
|
||||
private Tuple2<String, Map<String, Object>> buildDsl(final GdbElement element, final List<GdbElement.GdbProperty> properties,
|
||||
final boolean isVertex, final boolean firstAdd) {
|
||||
final Map<String, Object> params = new HashMap<>();
|
||||
final StringBuilder sb = new StringBuilder();
|
||||
if (isVertex) {
|
||||
sb.append(firstAdd ? ADD_V_START : UPDATE_V_START);
|
||||
} else {
|
||||
sb.append(firstAdd ? ADD_E_START : UPDATE_E_START);
|
||||
}
|
||||
if (firstAdd) {
|
||||
params.put(VAR_LABEL, element.getLabel());
|
||||
}
|
||||
params.put(VAR_ID, element.getId());
|
||||
|
||||
for (int i = 0; i < properties.size(); i++) {
|
||||
final GdbElement.GdbProperty prop = properties.get(i);
|
||||
return new Tuple2<>(dsl, params);
|
||||
}
|
||||
|
||||
sb.append(".property(");
|
||||
if (prop.getCardinality() == Key.PropertyType.set) {
|
||||
sb.append("set, ");
|
||||
}
|
||||
sb.append(VAR_PROP_KEY).append(i).append(", ").append(VAR_PROP_VALUE).append(i).append(")");
|
||||
private void setGraphDbElement(GdbElement element, Map<String, Object> properties,
|
||||
boolean isVertex, boolean firstAdd) throws GdbDuplicateIdException {
|
||||
int retry = 10;
|
||||
int idleTime = random.nextInt(10) + 10;
|
||||
Tuple2<String, Map<String, Object>> elementDsl = buildDsl(element, properties, isVertex, firstAdd);
|
||||
|
||||
params.put(VAR_PROP_KEY + i, prop.getKey());
|
||||
params.put(VAR_PROP_VALUE + i, prop.getValue());
|
||||
}
|
||||
|
||||
if (firstAdd) {
|
||||
params.put(VAR_LABEL, element.getLabel());
|
||||
if (!isVertex) {
|
||||
params.put(VAR_FROM, ((GdbEdge)element).getFrom());
|
||||
params.put(VAR_TO, ((GdbEdge)element).getTo());
|
||||
}
|
||||
}
|
||||
params.put(VAR_ID, element.getId());
|
||||
|
||||
return new Tuple2<>(sb.toString(), params);
|
||||
}
|
||||
|
||||
private void setGraphDbElement(final GdbElement element, final List<GdbElement.GdbProperty> properties, final boolean isVertex,
|
||||
final boolean firstAdd) throws GdbDuplicateIdException {
|
||||
int retry = 10;
|
||||
int idleTime = this.random.nextInt(10) + 10;
|
||||
final Tuple2<String, Map<String, Object>> elementDsl = buildDsl(element, properties, isVertex, firstAdd);
|
||||
|
||||
while (retry > 0) {
|
||||
try {
|
||||
runInternal(elementDsl.getFirst(), elementDsl.getSecond());
|
||||
log.debug("AddElement {}", element.getId());
|
||||
return;
|
||||
} catch (final Exception e) {
|
||||
final String cause = e.getCause() == null ? "" : e.getCause().toString();
|
||||
if (cause.contains("rejected from") || cause.contains("Timeout waiting to lock key")) {
|
||||
retry--;
|
||||
try {
|
||||
Thread.sleep(idleTime);
|
||||
} catch (final InterruptedException e1) {
|
||||
// ...
|
||||
}
|
||||
idleTime = Math.min(idleTime * 2, 2000);
|
||||
continue;
|
||||
} else if (firstAdd && cause.contains("GraphDB id exists")) {
|
||||
throw new GdbDuplicateIdException(e);
|
||||
}
|
||||
log.error("Add Failed id {}, dsl {}, params {}, e {}", element.getId(), elementDsl.getFirst(),
|
||||
elementDsl.getSecond(), e);
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
log.error("Add Failed id {}, dsl {}, params {}", element.getId(), elementDsl.getFirst(),
|
||||
elementDsl.getSecond());
|
||||
throw new RuntimeException("failed to queue new element to server");
|
||||
}
|
||||
while (retry > 0) {
|
||||
try {
|
||||
runInternal(elementDsl.getFirst(), elementDsl.getSecond());
|
||||
log.debug("AddElement {}", element.getId());
|
||||
return;
|
||||
} catch (Exception e) {
|
||||
String cause = e.getCause() == null ? "" : e.getCause().toString();
|
||||
if (cause.contains("rejected from")) {
|
||||
retry--;
|
||||
try {
|
||||
Thread.sleep(idleTime);
|
||||
} catch (InterruptedException e1) {
|
||||
// ...
|
||||
}
|
||||
idleTime = Math.min(idleTime * 2, 2000);
|
||||
continue;
|
||||
} else if (firstAdd && cause.contains("GraphDB id exists")) {
|
||||
throw new GdbDuplicateIdException(e);
|
||||
}
|
||||
log.error("Add Failed id {}, dsl {}, params {}, e {}", element.getId(),
|
||||
elementDsl.getFirst(), elementDsl.getSecond(), e);
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
log.error("Add Failed id {}, dsl {}, params {}", element.getId(),
|
||||
elementDsl.getFirst(), elementDsl.getSecond());
|
||||
throw new RuntimeException("failed to queue new element to server");
|
||||
}
|
||||
}
|
||||
|
||||
+36
-40
@@ -7,57 +7,53 @@ import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
|
||||
import com.alibaba.datax.common.exception.DataXException;
|
||||
import com.alibaba.datax.common.util.Configuration;
|
||||
import com.alibaba.datax.plugin.writer.gdbwriter.GdbWriterErrorCode;
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import com.alibaba.fastjson.JSONObject;
|
||||
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
|
||||
/**
|
||||
* @author jerrywang
|
||||
*
|
||||
*/
|
||||
public interface ConfigHelper {
|
||||
static void assertConfig(final String key, final Supplier<Boolean> f) {
|
||||
if (!f.get()) {
|
||||
throw DataXException.asDataXException(GdbWriterErrorCode.BAD_CONFIG_VALUE, key);
|
||||
}
|
||||
}
|
||||
static void assertConfig(String key, Supplier<Boolean> f) {
|
||||
if (!f.get()) {
|
||||
throw DataXException.asDataXException(GdbWriterErrorCode.BAD_CONFIG_VALUE, key);
|
||||
}
|
||||
}
|
||||
|
||||
static void assertHasContent(final Configuration config, final String key) {
|
||||
assertConfig(key, () -> StringUtils.isNotBlank(config.getString(key)));
|
||||
}
|
||||
static void assertHasContent(Configuration config, String key) {
|
||||
assertConfig(key, () -> StringUtils.isNotBlank(config.getString(key)));
|
||||
}
|
||||
|
||||
/**
|
||||
* NOTE: {@code Configuration::get(String, Class<T>)} doesn't work.
|
||||
*
|
||||
* @param conf
|
||||
* Configuration
|
||||
* @param key
|
||||
* key path to configuration
|
||||
* @param cls
|
||||
* Class of result type
|
||||
* @return the target configuration object of type T
|
||||
*/
|
||||
static <T> T getConfig(final Configuration conf, final String key, final Class<T> cls) {
|
||||
final JSONObject j = (JSONObject)conf.get(key);
|
||||
return JSON.toJavaObject(j, cls);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a configuration from the specified file on the classpath.
|
||||
*
|
||||
* @param name
|
||||
* file name
|
||||
* @return Configuration instance.
|
||||
*/
|
||||
static Configuration fromClasspath(final String name) {
|
||||
try (final InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream(name)) {
|
||||
return Configuration.from(is);
|
||||
} catch (final IOException e) {
|
||||
throw new IllegalArgumentException("File not found: " + name);
|
||||
}
|
||||
}
|
||||
/**
|
||||
* NOTE: {@code Configuration::get(String, Class<T>)} doesn't work.
|
||||
*
|
||||
* @param conf Configuration
|
||||
* @param key key path to configuration
|
||||
* @param cls Class of result type
|
||||
* @return the target configuration object of type T
|
||||
*/
|
||||
static <T> T getConfig(Configuration conf, String key, Class<T> cls) {
|
||||
JSONObject j = (JSONObject) conf.get(key);
|
||||
return JSON.toJavaObject(j, cls);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a configuration from the specified file on the classpath.
|
||||
*
|
||||
* @param name file name
|
||||
* @return Configuration instance.
|
||||
*/
|
||||
static Configuration fromClasspath(String name) {
|
||||
try (InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream(name)) {
|
||||
return Configuration.from(is);
|
||||
} catch (IOException e) {
|
||||
throw new IllegalArgumentException("File not found: " + name);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+10
-9
@@ -1,8 +1,9 @@
|
||||
/*
|
||||
* (C) 2019-present Alibaba Group Holding Limited.
|
||||
* (C) 2019-present Alibaba Group Holding Limited.
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public
|
||||
* License version 2 as published by the Free Software Foundation.
|
||||
* This program is free software; you can redistribute it and/or modify
|
||||
* it under the terms of the GNU General Public License version 2 as
|
||||
* published by the Free Software Foundation.
|
||||
*/
|
||||
package com.alibaba.datax.plugin.writer.gdbwriter.util;
|
||||
|
||||
@@ -12,11 +13,11 @@ package com.alibaba.datax.plugin.writer.gdbwriter.util;
|
||||
*/
|
||||
|
||||
public class GdbDuplicateIdException extends Exception {
|
||||
public GdbDuplicateIdException(Exception e) {
|
||||
super(e);
|
||||
}
|
||||
public GdbDuplicateIdException(Exception e) {
|
||||
super(e);
|
||||
}
|
||||
|
||||
public GdbDuplicateIdException() {
|
||||
super();
|
||||
}
|
||||
public GdbDuplicateIdException() {
|
||||
super();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -54,7 +54,7 @@
|
||||
<dependency>
|
||||
<groupId>org.apache.zookeeper</groupId>
|
||||
<artifactId>zookeeper</artifactId>
|
||||
<version>3.3.2</version>
|
||||
<version>3.4.14</version>
|
||||
</dependency>
|
||||
|
||||
|
||||
|
||||
@@ -132,7 +132,6 @@ MongoDBReader通过Datax框架从MongoDB并行的读取数据，通过主控的J
|
||||
* name：Column的名字。【必填】
|
||||
* type：Column的类型。【选填】
|
||||
* splitter：因为MongoDB支持数组类型，但是Datax框架本身不支持数组类型，所以mongoDB读出来的数组类型要通过这个分隔符合并成字符串。【选填】
|
||||
* query: MongoDB的额外查询条件。【选填】
|
||||
|
||||
#### 5 类型转换
|
||||
|
||||
|
||||
@@ -147,7 +147,7 @@ MysqlWriter 通过 DataX 框架获取 Reader 生成的协议数据，根据你
|
||||
|
||||
* **column**
|
||||
|
||||
* 描述：目的表需要写入数据的字段,字段之间用英文逗号分隔。例如: "column": ["id","name","age"]。如果要依次写入全部列，使用`*`表示, 例如: `"column": ["*"]`。
|
||||
* 描述：目的表需要写入数据的字段,字段之间用英文逗号分隔。例如: "column": ["id","name","age"]。如果要依次写入全部列，使用*表示, 例如: "column": ["*"]。
|
||||
|
||||
**column配置项必须指定，不能留空！**
|
||||
|
||||
|
||||
@@ -33,51 +33,47 @@ ODPSWriter插件用于实现往ODPS插入或者更新数据，主要提供给etl
|
||||
|
||||
```json
|
||||
{
|
||||
"job": {
|
||||
"setting": {
|
||||
"speed": {
|
||||
"byte": 1048576
|
||||
}
|
||||
},
|
||||
"content": [
|
||||
{
|
||||
"reader": {
|
||||
"name": "streamreader",
|
||||
"parameter": {
|
||||
"column": [
|
||||
{
|
||||
"value": "DataX",
|
||||
"type": "string"
|
||||
},
|
||||
{
|
||||
"value": "test",
|
||||
"type": "bytes"
|
||||
}
|
||||
],
|
||||
"sliceRecordCount": 100000
|
||||
}
|
||||
"job": {
|
||||
"setting": {
|
||||
"speed": {"byte": 1048576}
|
||||
},
|
||||
"writer": {
|
||||
"name": "odpswriter",
|
||||
"parameter": {
|
||||
"project": "chinan_test",
|
||||
"table": "odps_write_test00_partitioned",
|
||||
"partition": "school=SiChuan-School,class=1",
|
||||
"column": [
|
||||
"id",
|
||||
"name"
|
||||
],
|
||||
"accessId": "xxx",
|
||||
"accessKey": "xxxx",
|
||||
"truncate": true,
|
||||
"odpsServer": "http://sxxx/api",
|
||||
"tunnelServer": "http://xxx",
|
||||
"accountType": "aliyun"
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
"content": [
|
||||
{
|
||||
"reader": {
|
||||
"name": "streamreader",
|
||||
"parameter": {
|
||||
"column" : [
|
||||
{
|
||||
"value": "DataX",
|
||||
"type": "string"
|
||||
},
|
||||
{
|
||||
"value": "test",
|
||||
"type": "bytes"
|
||||
}
|
||||
],
|
||||
"sliceRecordCount": 100000
|
||||
}
|
||||
},
|
||||
"writer": {
|
||||
"name": "odpswriter",
|
||||
"parameter": {
|
||||
"project": "chinan_test",
|
||||
"table": "odps_write_test00_partitioned",
|
||||
"partition":"school=SiChuan-School,class=1",
|
||||
"column": ["id","name"],
|
||||
"accessId": "xxx",
|
||||
"accessKey": "xxxx",
|
||||
"truncate": true,
|
||||
"odpsServer": "http://sxxx/api",
|
||||
"tunnelServer": "http://xxx",
|
||||
"accountType": "aliyun"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
|
||||
@@ -101,7 +101,7 @@ OracleReader插件实现了从Oracle读取数据。在底层实现上，OracleRe
|
||||
"connection": [
|
||||
{
|
||||
"querySql": [
|
||||
"select db_id,on_line_flag from db_info where db_id < 10"
|
||||
"select db_id,on_line_flag from db_info where db_id < 10;"
|
||||
],
|
||||
"jdbcUrl": [
|
||||
"jdbc:oracle:thin:@[HOST_NAME]:PORT:[DATABASE_NAME]"
|
||||
|
||||
@@ -66,7 +66,6 @@
|
||||
<module>tsdbreader</module>
|
||||
<module>opentsdbreader</module>
|
||||
<module>cassandrareader</module>
|
||||
<module>gdbreader</module>
|
||||
|
||||
<!-- writer -->
|
||||
<module>mysqlwriter</module>
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
## Transformer定义
|
||||
|
||||
在数据同步、传输过程中，存在用户对于数据传输进行特殊定制化的需求场景，包括裁剪列、转换列等工作，可以借助ETL的T过程实现(Transformer)。DataX包含了完整的E(Extract)、T(Transformer)、L(Load)支持。
|
||||
在数据同步、传输过程中，存在用户对于数据传输进行特殊定制化的需求场景，包括裁剪列、转换列等工作，可以借助ETL的T过程实现(Transformer)。DataX包含了完成的E(Extract)、T(Transformer)、L(Load)支持。
|
||||
|
||||
## 运行模型
|
||||
|
||||
|
||||
@@ -187,6 +187,7 @@ TxtFileWriter实现了从DataX协议转为本地TXT文件功能，本地文件
|
||||
|
||||
| DataX 内部类型| 本地文件 数据类型 |
|
||||
| -------- | ----- |
|
||||
|
|
||||
| Long |Long |
|
||||
| Double |Double|
|
||||
| String |String|
|
||||
|
||||
Reference in New Issue
Block a user