Merge pull request #75 from datafibers-community/development_201612

Development 201612
datafibers-community · Dec 29, 2016 · 96e55e8 · 96e55e8
2 parents 79c3714 + 2c28d09
commit 96e55e8
Show file tree

Hide file tree

Showing 147 changed files with 11,383 additions and 1,285 deletions.
diff --git a/conf/connect-file-source.properties b/conf/connect-file-source.properties
diff --git a/conf/connect-standalone.properties b/conf/connect-standalone.properties
diff --git a/df-data-service.iml b/df-data-service.iml
@@ -12,20 +12,6 @@
     </content>
     <orderEntry type="inheritedJdk" />
     <orderEntry type="sourceFolder" forTests="false" />
-    <orderEntry type="library" name="Maven: org.apache.kafka:kafka_2.11:0.9.0.1" level="project" />
-    <orderEntry type="library" name="Maven: org.scala-lang.modules:scala-xml_2.11:1.0.4" level="project" />
-    <orderEntry type="library" name="Maven: net.sf.jopt-simple:jopt-simple:3.2" level="project" />
-    <orderEntry type="library" name="Maven: com.101tec:zkclient:0.8" level="project" />
-    <orderEntry type="library" name="Maven: org.scala-lang:scala-library:2.11.8" level="project" />
-    <orderEntry type="library" name="Maven: org.scala-lang.modules:scala-parser-combinators_2.11:1.0.4" level="project" />
-    <orderEntry type="library" name="Maven: commons-cli:commons-cli:1.3" level="project" />
-    <orderEntry type="library" name="Maven: org.apache.kafka:kafka_2.11:0.9.0.1" level="project" />
-    <orderEntry type="library" name="Maven: org.scala-lang.modules:scala-xml_2.11:1.0.4" level="project" />
-    <orderEntry type="library" name="Maven: net.sf.jopt-simple:jopt-simple:3.2" level="project" />
-    <orderEntry type="library" name="Maven: com.101tec:zkclient:0.8" level="project" />
-    <orderEntry type="library" name="Maven: org.scala-lang:scala-library:2.11.8" level="project" />
-    <orderEntry type="library" name="Maven: org.scala-lang.modules:scala-parser-combinators_2.11:1.0.4" level="project" />
-    <orderEntry type="library" name="Maven: commons-cli:commons-cli:1.3" level="project" />
     <orderEntry type="library" name="Maven: io.vertx:vertx-core:3.3.2" level="project" />
     <orderEntry type="library" name="Maven: io.netty:netty-common:4.1.1.Final" level="project" />
     <orderEntry type="library" name="Maven: io.netty:netty-buffer:4.1.1.Final" level="project" />
@@ -55,6 +41,7 @@
     <orderEntry type="library" name="Maven: org.joda:joda-convert:1.8.1" level="project" />
     <orderEntry type="library" name="Maven: com.github.mauricio:mysql-async_2.11:0.2.20" level="project" />
     <orderEntry type="library" name="Maven: io.vertx:vertx-sql-common:3.3.2" level="project" />
+    <orderEntry type="library" name="Maven: org.mongodb:mongo-java-driver:3.3.0" level="project" />
     <orderEntry type="library" name="Maven: org.apache.flink:flink-table_2.11:1.1.2" level="project" />
     <orderEntry type="library" name="Maven: org.apache.flink:flink-streaming-scala_2.11:1.1.2" level="project" />
     <orderEntry type="library" name="Maven: org.apache.flink:flink-scala_2.11:1.1.2" level="project" />

diff --git a/pom.xml b/pom.xml
@@ -12,7 +12,7 @@
         <vertx.version>3.3.2</vertx.version>
         <vertx.rest.client.version>2.1.0</vertx.rest.client.version>
         <flink.version>1.1.2</flink.version>
-        <kafka.version>0.10.0.0</kafka.version>
+        <kafka.version>0.10.0.1</kafka.version>
         <zookeeperclinet.version>0.7</zookeeperclinet.version>
         <hivejdbc.version>2.1.0</hivejdbc.version>
         <kafka.topic.manager.version>0.9.0.1</kafka.topic.manager.version>
@@ -166,11 +166,11 @@
         </dependency>
 
         <!-- Logging -->
-        <dependency>
+<!--        <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-log4j12</artifactId>
             <version>1.7.7</version>
-        </dependency>
+        </dependency>-->
 
         <!-- For the tests -->
         <dependency>

diff --git a/src/main/conf/app_conf.json b/src/main/conf/app_conf.json
@@ -5,6 +5,7 @@
 
   "db.name":"df",
   "db.collection.name":"df_processor",
+  "db.metadata.collection.name":"df_meta",
   "repo.connection.string":"mongodb://localhost:27017",
 
   "kafka.server.port":9092,
@@ -14,6 +15,7 @@
   "kafka.connect.rest.host":"localhost",
   "kafka.connect.enable":"true",
   "kafka.connect.import.start":"true",
+  "kafka.topic.df.metadata":"df_meta",
 
   "transform.engine.flink.enable":"true",
   "flink.servers.host":"localhost",
@@ -22,5 +24,7 @@
   "flink.trans.client.timeout":120000,
 
   "zookeeper.server.host":"localhost",
-  "zookeeper.server.port":2181
+  "zookeeper.server.port":2181,
+
+  "df.metadata.schema":"{\"type\":\"record\",\"name\": \"df_meta\",\"fields\":[{\"name\": \"cuid\", \"type\": \"string\"},{\"name\": \"file_name\", \"type\": \"string\"},{\"name\": \"file_size\", \"type\": \"string\"}, {\"name\": \"file_owner\", \"type\": \"string\"},{\"name\": \"last_modified_timestamp\", \"type\": \"string\"},{\"name\": \"current_timestamp\", \"type\": \"string\"},{\"name\": \"current_timemillis\", \"type\": \"long\"},{\"name\": \"stream_offset\", \"type\": \"string\"},{\"name\": \"topic_sent\", \"type\": \"string\"},{\"name\": \"schema_subject\", \"type\": \"string\"},{\"name\": \"schema_version\", \"type\": \"string\"},{\"name\": \"status\", \"type\": \"string\"}]}"
 }
diff --git a/src/main/init/df_installed.json b/src/main/init/df_installed.json
@@ -0,0 +1,10 @@
+{"_id":{"$oid":"5852fe103ea0bf9558bd9bea"},"class":"io.confluent.connect.jdbc.JdbcSourceConnector","name":"JDBC Source Connector","type":"Kafka","subtype":"Source","certified":"datafibers","pubisher":"confluent","link":"http://www.confluent.io","version":"0.1.0","config":null}
+{"_id":{"$oid":"5852fe903ea0bf9558bd9beb"},"class":"com.datafibers.kafka.connect.FileGenericSourceConnector","name":"FileGeneric Source Connector","type":"Kafka","subtype":"Source","certified":"datafibers","pubisher":"datafibers","link":"https://github.com/datafibers-community/df_certified_connects","version":"0.1.0","config":null}
+{"_id":{"$oid":"5852fff33ea0bf9558bd9bec"},"class":"com.datafibers.kafka.connect.SchemaedFileSinkConnector","name":"SchemaedFile Sink Connector","type":"Kafka","subtype":"Sink","certified":"to be certified","pubisher":"dbtucker","link":"https://github.com/dbtucker/kafka-connect-schemaedfile","version":"0.1.0","config":null}
+{"_id":{"$oid":"585300213ea0bf9558bd9bed"},"class":"com.datafibers.kafka.connect.SchemaedFileSourceConnector","name":"SchemaedFile Source Connector","type":"Kafka","subtype":"Source","certified":"to be certified","pubisher":"dbtucker","link":"https://github.com/dbtucker/kafka-connect-schemaedfile","version":"0.1.0","config":null}
+{"_id":{"$oid":"585300ce3ea0bf9558bd9bee"},"class":"io.confluent.connect.hdfs.HdfsSinkConnector","name":"HDFS Sink Connector","type":"Kafka","subtype":"Sink","certified":"to be certified","pubisher":"confluent","link":"http://www.confluent.io","version":"0.1.0","config":null}
+{"_id":{"$oid":"585301003ea0bf9558bd9bef"},"class":"io.confluent.connect.hdfs.tools.SchemaSourceConnector","name":"HDFS Source Connector","type":"Kafka","subtype":"Source","certified":"to be certified","pubisher":"confluent","link":"http://www.confluent.io","version":"0.1.0","config":null}
+{"_id":{"$oid":"5853013a3ea0bf9558bd9bf0"},"class":"org.apache.kafka.connect.file.FileStreamSinkConnector","name":"FileStream Sink Connector","type":"Kafka","subtype":"Sink","certified":"to be certified","pubisher":"confluent","link":"http://www.confluent.io","version":"0.1.0","config":null}
+{"_id":{"$oid":"585301523ea0bf9558bd9bf1"},"class":"org.apache.kafka.connect.file.FileStreamSourceConnector","name":"FileStream Source Connector","type":"Kafka","subtype":"Source","certified":"to be certified","pubisher":"confluent","link":"http://www.confluent.io","version":"0.1.0","config":null}
+{"_id":{"$oid":"5853052a3ea0bf9558bd9bf2"},"class":"org.apache.kafka.connect.mongodb.MongodbSinkConnector","name":"Mongodb Sink Connector","type":"Kafka","subtype":"Sink","certified":"datafibers","pubisher":"datareply","link":"https://github.com/DataReply/kafka-connect-mongodb","version":"0.1.0","config":null}
+{"_id":{"$oid":"585305503ea0bf9558bd9bf3"},"class":"org.apache.kafka.connect.mongodb.MongodbSourceConnector","name":"Mongodb Source Connector","type":"Kafka","subtype":"Source","certified":"datafibers","pubisher":"datareply","link":"https://github.com/DataReply/kafka-connect-mongodb","version":"0.1.0","config":null}
diff --git a/src/main/init/import_meta.sh b/src/main/init/import_meta.sh
@@ -0,0 +1,4 @@
+#!/bin/bash
+set -e
+
+mongoimport -c df_installed -d DEFAULT_DB --file df_installed.json
diff --git a/src/main/java/com/datafibers/flinknext/AvroDeserializationSchema.java b/src/main/java/com/datafibers/flinknext/AvroDeserializationSchema.java
@@ -12,19 +12,11 @@
 import org.apache.flink.api.java.typeutils.TypeExtractor;
 import org.apache.flink.streaming.util.serialization.DeserializationSchema;
 import org.apache.kafka.common.errors.SerializationException;
-import org.codehaus.jackson.JsonNode;
-import org.codehaus.jackson.map.ObjectMapper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.net.URL;
+import org.apache.log4j.Logger;
 import java.nio.ByteBuffer;
 
 public class AvroDeserializationSchema implements DeserializationSchema<GenericRecord> {
-    private static final Logger LOG = LoggerFactory.getLogger(AvroDeserializationSchema.class);
+    private static final Logger LOG = Logger.getLogger(AvroDeserializationSchema.class);
 
     private static final long serialVersionUID = 4330538776656642778L;
 

diff --git a/src/main/java/com/datafibers/flinknext/AvroRowDeserializationSchema.java b/src/main/java/com/datafibers/flinknext/AvroRowDeserializationSchema.java
@@ -15,10 +15,6 @@
 import org.apache.flink.api.table.typeutils.RowTypeInfo;
 import org.apache.flink.streaming.util.serialization.DeserializationSchema;
 import org.apache.flink.util.Preconditions;
-import org.apache.kafka.common.errors.SerializationException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Properties;
@@ -32,7 +28,7 @@
  * <p>Failure during deserialization are forwarded as wrapped IOExceptions.
  */
 public class AvroRowDeserializationSchema implements DeserializationSchema<Row> {
-    private static final Logger LOG = LoggerFactory.getLogger(AvroRowDeserializationSchema.class);
+
     private static final long serialVersionUID = 4330538776656642779L;
 
     /** Field names to parse. Indices match fieldTypes indices. */