Skip to content

Commit 3207b63

Browse files
FuYouJFuYouJ
committed
对example工程结构进行了重新规划,给出两个编写测试模块的案例。
对example工程结构进行了重新规划,给出两个编写测试模块的案例。
1 parent c0511ba commit 3207b63

File tree

24 files changed

+336
-304
lines changed

24 files changed

+336
-304
lines changed

‎README.md

+1-4
Original file line numberDiff line numberDiff line change
@@ -100,10 +100,7 @@ DataX目前已经有了比较全面的插件体系，主流的RDBMS数据库、N
100100
- 整库迁移：https://help.aliyun.com/document_detail/137809.html
101101
- 批量上云：https://help.aliyun.com/document_detail/146671.html
102102
- 更新更多能力请访问：https://help.aliyun.com/document_detail/137663.html
103-
-
104-
# 本地快速调试读写插件定位BUG
105-
在example模块可以方便快速的在本地运行任务
106-
点击：[datax-example使用](https://github.com/alibaba/DataX/datax-example/doc/README.md)
103+
-
107104

108105
# 我要开发新的插件
109106

‎datax-example/datax-example-core/pom.xml

+20
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns="http://maven.apache.org/POM/4.0.0"
3+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5+
<modelVersion>4.0.0</modelVersion>
6+
<parent>
7+
<groupId>com.alibaba.datax</groupId>
8+
<artifactId>datax-example</artifactId>
9+
<version>0.0.1-SNAPSHOT</version>
10+
</parent>
11+
12+
<artifactId>datax-example-core</artifactId>
13+
14+
<properties>
15+
<maven.compiler.source>8</maven.compiler.source>
16+
<maven.compiler.target>8</maven.compiler.target>
17+
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
18+
</properties>
19+
20+
</project>

‎datax-example/src/main/java/com/alibaba/datax/example/ExampleContainer.java ‎datax-example/datax-example-core/src/main/java/com/alibaba/datax/example/ExampleContainer.java

File renamed without changes.

‎datax-example/src/main/java/com/alibaba/datax/example/Main.java ‎datax-example/datax-example-core/src/main/java/com/alibaba/datax/example/Main.java

File renamed without changes.

‎datax-example/src/main/java/com/alibaba/datax/example/util/ExampleConfigParser.java ‎datax-example/datax-example-core/src/main/java/com/alibaba/datax/example/util/ExampleConfigParser.java

+2-3
Original file line numberDiff line numberDiff line change
@@ -63,9 +63,8 @@ private static Configuration parsePluginsConfig(Map<String, String> pluginTypeMa
6363
" <include>**/*.*</include>\n" +
6464
" </includes>\n" +
6565
" <filtering>true</filtering>\n" +
66-
" </resource>\n [可参阅streamreader pom文件] \n" +
67-
"3：If you are using 'datax-example' as the startup module, " +
68-
"check whether the 'example' module has imported the dependencies of the plugin. Refer to the 'pom' file of the 'example' module";
66+
" </resource>\n [Refer to the streamreader pom file] \n" +
67+
"3: Check that the datax-yourPlugin-example module imported your test plugin";
6968
message = String.format(message, failedPlugin);
7069
throw DataXException.asDataXException(FrameworkErrorCode.PLUGIN_INIT_ERROR, message);
7170
}

‎datax-example/src/main/java/com/alibaba/datax/example/util/PathUtil.java ‎datax-example/datax-example-core/src/main/java/com/alibaba/datax/example/util/PathUtil.java

File renamed without changes.

‎datax-example/src/main/resources/example/conf/core.json ‎datax-example/datax-example-core/src/main/resources/example/conf/core.json

File renamed without changes.

‎datax-example/datax-example-core/src/test/java/com/alibaba/datax/example/util/PathUtilTest.java

+19
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
package com.alibaba.datax.example.util;
2+
3+
import org.junit.Assert;
4+
import org.junit.Test;
5+
6+
/**
7+
* {@code Author} FuYouJ
8+
* {@code Date} 2023/8/19 21:38
9+
*/
10+
11+
public class PathUtilTest {
12+
13+
@Test
14+
public void testParseClassPathFile() {
15+
String path = "/pathTest.json";
16+
String absolutePathFromClassPath = PathUtil.getAbsolutePathFromClassPath(path);
17+
Assert.assertNotNull(absolutePathFromClassPath);
18+
}
19+
}

‎datax-example/datax-example-core/src/test/resources/pathTest.json

+1
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
{}

‎datax-example/datax-example-neo4j/pom.xml

+43
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns="http://maven.apache.org/POM/4.0.0"
3+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5+
<modelVersion>4.0.0</modelVersion>
6+
<parent>
7+
<groupId>com.alibaba.datax</groupId>
8+
<artifactId>datax-example</artifactId>
9+
<version>0.0.1-SNAPSHOT</version>
10+
</parent>
11+
12+
<artifactId>datax-example-neo4j</artifactId>
13+
14+
<properties>
15+
<maven.compiler.source>8</maven.compiler.source>
16+
<maven.compiler.target>8</maven.compiler.target>
17+
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
18+
<test.container.version>1.17.6</test.container.version>
19+
<neo4j-java-driver.version>4.4.9</neo4j-java-driver.version>
20+
</properties>
21+
<dependencies>
22+
<dependency>
23+
<groupId>com.alibaba.datax</groupId>
24+
<artifactId>datax-example-core</artifactId>
25+
<version>0.0.1-SNAPSHOT</version>
26+
</dependency>
27+
<dependency>
28+
<groupId>org.testcontainers</groupId>
29+
<artifactId>testcontainers</artifactId>
30+
<version>${test.container.version}</version>
31+
</dependency>
32+
<dependency>
33+
<groupId>com.alibaba.datax</groupId>
34+
<artifactId>neo4jwriter</artifactId>
35+
<version>0.0.1-SNAPSHOT</version>
36+
</dependency>
37+
<dependency>
38+
<groupId>com.alibaba.datax</groupId>
39+
<artifactId>datax-example-streamreader</artifactId>
40+
<version>0.0.1-SNAPSHOT</version>
41+
</dependency>
42+
</dependencies>
43+
</project>

‎datax-example/datax-example-neo4j/src/test/java/com/alibaba/datax/example/neo4j/StreamReader2Neo4jWriterTest.java

+138
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,138 @@
1+
package com.alibaba.datax.example.neo4j;
2+
3+
import com.alibaba.datax.example.ExampleContainer;
4+
import com.alibaba.datax.example.util.PathUtil;
5+
import org.junit.After;
6+
import org.junit.Assert;
7+
import org.junit.Before;
8+
import org.junit.Test;
9+
import org.neo4j.driver.*;
10+
import org.neo4j.driver.types.Node;
11+
import org.slf4j.Logger;
12+
import org.slf4j.LoggerFactory;
13+
import org.testcontainers.containers.GenericContainer;
14+
import org.testcontainers.containers.Network;
15+
import org.testcontainers.containers.output.Slf4jLogConsumer;
16+
import org.testcontainers.lifecycle.Startables;
17+
import org.testcontainers.shaded.org.awaitility.Awaitility;
18+
import org.testcontainers.utility.DockerImageName;
19+
import org.testcontainers.utility.DockerLoggerFactory;
20+
21+
import java.net.URI;
22+
import java.util.Arrays;
23+
import java.util.concurrent.TimeUnit;
24+
import java.util.stream.Stream;
25+
26+
/**
27+
* {@code Author} FuYouJ
28+
* {@code Date} 2023/8/19 21:48
29+
*/
30+
31+
public class StreamReader2Neo4jWriterTest {
32+
private static final Logger LOGGER = LoggerFactory.getLogger(StreamReader2Neo4jWriterTest.class);
33+
private static final String CONTAINER_IMAGE = "neo4j:5.9.0";
34+
35+
private static final String CONTAINER_HOST = "neo4j-host";
36+
private static final int HTTP_PORT = 7474;
37+
private static final int BOLT_PORT = 7687;
38+
private static final String CONTAINER_NEO4J_USERNAME = "neo4j";
39+
private static final String CONTAINER_NEO4J_PASSWORD = "Test@12343";
40+
private static final URI CONTAINER_URI = URI.create("neo4j://localhost:" + BOLT_PORT);
41+
42+
protected static final Network NETWORK = Network.newNetwork();
43+
44+
private GenericContainer<?> container;
45+
protected Driver neo4jDriver;
46+
protected Session neo4jSession;
47+
private static final int CHANNEL = 5;
48+
private static final int READER_NUM = 10;
49+
50+
@Before
51+
public void init() {
52+
DockerImageName imageName = DockerImageName.parse(CONTAINER_IMAGE);
53+
container =
54+
new GenericContainer<>(imageName)
55+
.withNetwork(NETWORK)
56+
.withNetworkAliases(CONTAINER_HOST)
57+
.withExposedPorts(HTTP_PORT, BOLT_PORT)
58+
.withEnv(
59+
"NEO4J_AUTH",
60+
CONTAINER_NEO4J_USERNAME + "/" + CONTAINER_NEO4J_PASSWORD)
61+
.withEnv("apoc.export.file.enabled", "true")
62+
.withEnv("apoc.import.file.enabled", "true")
63+
.withEnv("apoc.import.file.use_neo4j_config", "true")
64+
.withEnv("NEO4J_PLUGINS", "[\"apoc\"]")
65+
.withLogConsumer(
66+
new Slf4jLogConsumer(
67+
DockerLoggerFactory.getLogger(CONTAINER_IMAGE)));
68+
container.setPortBindings(
69+
Arrays.asList(
70+
String.format("%s:%s", HTTP_PORT, HTTP_PORT),
71+
String.format("%s:%s", BOLT_PORT, BOLT_PORT)));
72+
Startables.deepStart(Stream.of(container)).join();
73+
LOGGER.info("container started");
74+
Awaitility.given()
75+
.ignoreExceptions()
76+
.await()
77+
.atMost(30, TimeUnit.SECONDS)
78+
.untilAsserted(this::initConnection);
79+
}
80+
81+
//在neo4jWriter模块使用Example测试整个job,方便发现整个流程的代码问题
82+
@Test
83+
public void streamReader2Neo4j() {
84+
85+
deleteHistoryIfExist();
86+
87+
String path = "/streamreader2neo4j.json";
88+
String jobPath = PathUtil.getAbsolutePathFromClassPath(path);
89+
90+
ExampleContainer.start(jobPath);
91+
92+
//根据channel和reader的mock数据，校验结果集是否符合预期
93+
verifyWriteResult();
94+
}
95+
96+
private void deleteHistoryIfExist() {
97+
String query = "match (n:StreamReader) return n limit 1";
98+
String delete = "match (n:StreamReader) delete n";
99+
if (neo4jSession.run(query).hasNext()) {
100+
neo4jSession.run(delete);
101+
}
102+
}
103+
104+
private void verifyWriteResult() {
105+
int total = CHANNEL * READER_NUM;
106+
String query = "match (n:StreamReader) return n";
107+
Result run = neo4jSession.run(query);
108+
int count = 0;
109+
while (run.hasNext()) {
110+
Record record = run.next();
111+
Node node = record.get("n").asNode();
112+
if (node.hasLabel("StreamReader")) {
113+
count++;
114+
}
115+
}
116+
Assert.assertEquals(count, total);
117+
}
118+
@After
119+
public void destroy() {
120+
if (neo4jSession != null) {
121+
neo4jSession.close();
122+
}
123+
if (neo4jDriver != null) {
124+
neo4jDriver.close();
125+
}
126+
if (container != null) {
127+
container.close();
128+
}
129+
}
130+
131+
private void initConnection() {
132+
neo4jDriver =
133+
GraphDatabase.driver(
134+
CONTAINER_URI,
135+
AuthTokens.basic(CONTAINER_NEO4J_USERNAME, CONTAINER_NEO4J_PASSWORD));
136+
neo4jSession = neo4jDriver.session(SessionConfig.forDatabase("neo4j"));
137+
}
138+
}

‎datax-example/datax-example-neo4j/src/test/resources/streamreader2neo4j.json

+51
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
{
2+
"job": {
3+
"content": [
4+
{
5+
"reader": {
6+
"name": "streamreader",
7+
"parameter": {
8+
"sliceRecordCount": 10,
9+
"column": [
10+
{
11+
"type": "string",
12+
"value": "StreamReader"
13+
},
14+
{
15+
"type": "string",
16+
"value": "1997"
17+
}
18+
]
19+
}
20+
},
21+
"writer": {
22+
"name": "neo4jWriter",
23+
"parameter": {
24+
"uri": "bolt://localhost:7687",
25+
"username":"neo4j",
26+
"password":"Test@12343",
27+
"database":"neo4j",
28+
"cypher": "unwind $batch as row CALL apoc.cypher.doIt( 'create (n:`' + row.Label + '`{id:$id})' ,{id: row.id} ) YIELD value RETURN 1 ",
29+
"batchDataVariableName": "batch",
30+
"batchSize": "3",
31+
"properties": [
32+
{
33+
"name": "Label",
34+
"type": "string"
35+
},
36+
{
37+
"name": "id",
38+
"type": "STRING"
39+
}
40+
]
41+
}
42+
}
43+
}
44+
],
45+
"setting": {
46+
"speed": {
47+
"channel": 5
48+
}
49+
}
50+
}
51+
}

‎datax-example/datax-example-streamreader/pom.xml

+37
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns="http://maven.apache.org/POM/4.0.0"
3+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5+
<modelVersion>4.0.0</modelVersion>
6+
<parent>
7+
<groupId>com.alibaba.datax</groupId>
8+
<artifactId>datax-example</artifactId>
9+
<version>0.0.1-SNAPSHOT</version>
10+
</parent>
11+
12+
<artifactId>datax-example-streamreader</artifactId>
13+
14+
<properties>
15+
<maven.compiler.source>8</maven.compiler.source>
16+
<maven.compiler.target>8</maven.compiler.target>
17+
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
18+
</properties>
19+
<dependencies>
20+
<dependency>
21+
<groupId>com.alibaba.datax</groupId>
22+
<artifactId>datax-example-core</artifactId>
23+
<version>0.0.1-SNAPSHOT</version>
24+
</dependency>
25+
<dependency>
26+
<groupId>com.alibaba.datax</groupId>
27+
<artifactId>streamreader</artifactId>
28+
<version>0.0.1-SNAPSHOT</version>
29+
</dependency>
30+
<dependency>
31+
<groupId>com.alibaba.datax</groupId>
32+
<artifactId>streamwriter</artifactId>
33+
<version>0.0.1-SNAPSHOT</version>
34+
</dependency>
35+
</dependencies>
36+
37+
</project>

‎datax-example/src/test/java/com/alibaba/datax/example/streamreader/StreamReader2StreamWriterTest.java ‎datax-example/datax-example-streamreader/src/test/java/com/alibaba/datax/example/streamreader/StreamReader2StreamWriterTest.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@
1212
public class StreamReader2StreamWriterTest {
1313
@Test
1414
public void testStreamReader2StreamWriter() {
15-
String path = "/job/stream/stream2stream.json";
15+
String path = "/stream2stream.json";
1616
String jobPath = PathUtil.getAbsolutePathFromClassPath(path);
1717
ExampleContainer.start(jobPath);
1818
}

‎datax-example/src/main/resources/job/stream2stream.json ‎datax-example/datax-example-streamreader/src/test/resources/stream2stream.json

File renamed without changes.

0 commit comments

Comments
 (0)