Add milvusReader and milvusWriter to support read and write data for milvus

milvusreader/pom.xml

@@ -0,0 +1,109 @@
+<?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">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>com.alibaba.datax</groupId>
+        <artifactId>datax-all</artifactId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>milvusreader</artifactId>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <kotlin.code.style>official</kotlin.code.style>
+        <kotlin.compiler.jvmTarget>1.8</kotlin.compiler.jvmTarget>
+    </properties>
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <artifactId>guava</artifactId>
+                <groupId>com.google.guava</groupId>
+                <version>32.0.1-jre</version>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.milvus</groupId>
+            <artifactId>milvus-sdk-java</artifactId>
+            <version>2.4.8</version>
+        </dependency>
+        <dependency>
+            <groupId>org.jetbrains.kotlin</groupId>
+            <artifactId>kotlin-test-junit5</artifactId>
+            <version>2.0.0</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <version>5.10.0</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.jetbrains.kotlin</groupId>
+            <artifactId>kotlin-stdlib</artifactId>
+            <version>2.0.0</version>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba.datax</groupId>
+            <artifactId>datax-common</artifactId>
+            <version>0.0.1-SNAPSHOT</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>1.18.30</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <resources>
+            <!--将resource目录也输出到target-->
+            <resource>
+                <directory>src/main/resources</directory>
+                <includes>
+                    <include>**/*.*</include>
+                </includes>
+                <filtering>true</filtering>
+            </resource>
+        </resources>
+        <plugins>
+            <!-- compiler plugin -->
+            <plugin>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>${jdk-version}</source>
+                    <target>${jdk-version}</target>
+                    <encoding>${project-sourceEncoding}</encoding>
+                </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>
+        </plugins>
+    </build>
+
+</project>

milvusreader/src/main/assembly/package.xml

@@ -0,0 +1,36 @@
+<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/milvusreader</outputDirectory>
+        </fileSet>
+        <fileSet>
+            <directory>target/</directory>
+            <includes>
+                <include>milvuswriter-0.0.1-SNAPSHOT.jar</include>
+            </includes>
+            <outputDirectory>plugin/reader/milvusreader</outputDirectory>
+        </fileSet>
+    </fileSets>
+
+    <dependencySets>
+        <dependencySet>
+            <useProjectArtifact>false</useProjectArtifact>
+            <outputDirectory>plugin/reader/milvusreader/libs</outputDirectory>
+            <scope>runtime</scope>
+        </dependencySet>
+    </dependencySets>
+</assembly>

milvusreader/src/main/doc/milvusreader.md

@@ -0,0 +1,105 @@
+### Datax MilvusReader
+#### 1 快速介绍
+
+MilvusReader 插件利用 Milvus 的java客户端MilvusClient进行Milvus的读操作。
+
+#### 2 实现原理
+
+MilvusReader通过Datax框架从Milvus读取数据，通过主控的JOB程序按照指定的规则对Milvus中的数据进行分片，并行读取，然后将Milvus支持的类型通过逐一判断转换成Datax支持的类型。
+
+#### 3 功能说明
+* 该示例从Milvus读一份Collection数据到另一个Milvus。
+```json
+{
+  "job": {
+    "content": [
+      {
+        "reader": {
+          "name": "milvusreader",
+          "parameter": {
+            "uri": "https://****.aws-us-west-2.vectordb.zillizcloud.com:19532",
+            "token": "*****",
+            "collection": "medium_articles",
+            "batchSize": 10
+          }
+        },
+        "writer": {
+          "name": "milvuswriter",
+          "parameter": {
+            "uri": "https://*****.aws-us-west-2.vectordb.zillizcloud.com:19530",
+            "token": "*****",
+            "collection": "medium_articles",
+            "batchSize": 10,
+            "column": [
+              {
+                "name": "id",
+                "type": "Int64",
+                "isPrimaryKey": true
+              },
+              {
+                "name": "title_vector",
+                "type": "FloatVector",
+                "dimension": 768
+              },
+              {
+                "name": "title",
+                "type": "VarChar",
+                "maxLength": 1000
+              },
+              {
+                "name": "link",
+                "type": "VarChar",
+                "maxLength": 1000
+              },
+              {
+                "name": "reading_time",
+                "type": "Int64"
+              },
+              {
+                "name": "publication",
+                "type": "VarChar",
+                "maxLength": 1000
+              },
+              {
+                "name": "claps",
+                "type": "Int64"
+              },
+              {
+                "name": "responses",
+                "type": "Int64"
+              }
+            ]
+          }
+        }
+      }
+    ],
+    "setting": {
+      "speed": {
+        "channel": 1
+      }
+    }
+  }
+}
+```
+
+#### 4 参数说明
+
+* uri： Milvus Cluster endpoint。【必填】
+* token：Milvus的连接token。【必填】
+* collection： 读取数据的collection。【必填】
+* partition： 读取数据的partition。【选填】
+* batchSize: 每次读取数据的行数【选填】
+
+#### 5 类型转换
+
+| DataX 内部类型| Milvus 数据类型     |
+| -------- |-----------------|
+| Long     | int             |
+| Double   | double          |
+| String   | string, varchar |
+| Boolean  | bool            |
+
+- 当前暂不支持读取dynamic schema的数据，及按partition读取
+
+#### 6 性能报告
+#### 7 测试报告

milvusreader/src/main/java/com/alibaba/datax/plugin/reader/milvusreader/BufferUtils.java

@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.datax.plugin.reader.milvusreader;
+
+import java.nio.Buffer;
+import java.nio.ByteBuffer;
+
+public class BufferUtils {
+
+    public static ByteBuffer toByteBuffer(Short[] shortArray) {
+        ByteBuffer byteBuffer = ByteBuffer.allocate(shortArray.length * 2);
+
+        for (Short value : shortArray) {
+            byteBuffer.putShort(value);
+        }
+
+        // Compatible compilation and running versions are not consistent
+        // Flip the buffer to prepare for reading
+        ((Buffer) byteBuffer).flip();
+
+        return byteBuffer;
+    }
+
+    public static Short[] toShortArray(ByteBuffer byteBuffer) {
+        Short[] shortArray = new Short[byteBuffer.capacity() / 2];
+
+        for (int i = 0; i < shortArray.length; i++) {
+            shortArray[i] = byteBuffer.getShort();
+        }
+
+        return shortArray;
+    }
+
+    public static ByteBuffer toByteBuffer(Float[] floatArray) {
+        ByteBuffer byteBuffer = ByteBuffer.allocate(floatArray.length * 4);
+
+        for (float value : floatArray) {
+            byteBuffer.putFloat(value);
+        }
+
+        ((Buffer) byteBuffer).flip();
+
+        return byteBuffer;
+    }
+
+    public static Float[] toFloatArray(ByteBuffer byteBuffer) {
+        Float[] floatArray = new Float[byteBuffer.capacity() / 4];
+
+        for (int i = 0; i < floatArray.length; i++) {
+            floatArray[i] = byteBuffer.getFloat();
+        }
+
+        return floatArray;
+    }
+
+    public static ByteBuffer toByteBuffer(Double[] doubleArray) {
+        ByteBuffer byteBuffer = ByteBuffer.allocate(doubleArray.length * 8);
+
+        for (double value : doubleArray) {
+            byteBuffer.putDouble(value);
+        }
+
+        ((Buffer) byteBuffer).flip();
+
+        return byteBuffer;
+    }
+
+    public static Double[] toDoubleArray(ByteBuffer byteBuffer) {
+        Double[] doubleArray = new Double[byteBuffer.capacity() / 8];
+
+        for (int i = 0; i < doubleArray.length; i++) {
+            doubleArray[i] = byteBuffer.getDouble();
+        }
+
+        return doubleArray;
+    }
+
+    public static ByteBuffer toByteBuffer(Integer[] intArray) {
+        ByteBuffer byteBuffer = ByteBuffer.allocate(intArray.length * 4);
+
+        for (int value : intArray) {
+            byteBuffer.putInt(value);
+        }
+
+        ((Buffer) byteBuffer).flip();
+
+        return byteBuffer;
+    }
+
+    public static Integer[] toIntArray(ByteBuffer byteBuffer) {
+        Integer[] intArray = new Integer[byteBuffer.capacity() / 4];
+
+        for (int i = 0; i < intArray.length; i++) {
+            intArray[i] = byteBuffer.getInt();
+        }
+
+        return intArray;
+    }
+}

milvusreader/src/main/java/com/alibaba/datax/plugin/reader/milvusreader/KeyConstant.java

@@ -0,0 +1,10 @@
+package com.alibaba.datax.plugin.reader.milvusreader;
+
+public class KeyConstant {
+    public static final String URI = "uri";
+    public static final String TOKEN = "token";
+    public static final String DATABASE = "database";
+    public static final String COLLECTION = "collection";
+    public static final String PARTITION = "partition";
+    public static final String BATCH_SIZE = "batchSize";
+}