Kid xiong.fea.input.hive

build.sbt

@@ -12,12 +12,11 @@ assemblyMergeStrategy in assembly := {
   case PathList("com", "esotericsoftware", xs @ _*) => MergeStrategy.last
   case PathList("com", "codahale", xs @ _*) => MergeStrategy.last
   case PathList("com", "yammer", xs @ _*) => MergeStrategy.last
-  case "about.html" => MergeStrategy.rename
-  case "META-INF/ECLIPSEF.RSA" => MergeStrategy.last
-  case "META-INF/mailcap" => MergeStrategy.last
-  case "META-INF/mimetypes.default" => MergeStrategy.last
   case PathList(ps @ _*) if ps.last endsWith ".html" => MergeStrategy.first
+  case PathList(ps @ _*) if ps.last endsWith ".xml" => MergeStrategy.first
   case PathList(ps @ _*) if ps.last endsWith ".class" => MergeStrategy.first
+  case PathList(ps @ _*) if ps.last endsWith ".thrift" => MergeStrategy.first
+  case "UnusedStubClass.class" => MergeStrategy.last
   case x =>
     val oldStrategy = (assemblyMergeStrategy in assembly).value
     oldStrategy(x)

docs/zh-cn/configuration/_sidebar.md

@@ -15,6 +15,8 @@
     - [Hdfs](/zh-cn/configuration/input-plugins/Hdfs)
     - [HdfsStream](/zh-cn/configuration/input-plugins/HdfsStream)
     - [KafkaStream](/zh-cn/configuration/input-plugins/KafkaStream)
+    - [Kudu](/zh-cn/configuration/input-plugins/Kudu)
+    - [MongoDB](/zh-cn/configuration/input-plugins/MongoDB)
     - [S3Stream](/zh-cn/configuration/input-plugins/S3Stream)
     - [SocketStream](/zh-cn/configuration/input-plugins/SocketStream)
 
@@ -49,6 +51,8 @@
     - [Hdfs](/zh-cn/configuration/output-plugins/Hdfs)
     - [Jdbc](/zh-cn/configuration/output-plugins/Jdbc)
     - [Kafka](/zh-cn/configuration/output-plugins/Kafka)
+    - [Kudu](/zh-cn/configuration/output-plugins/Kudu)
+    - [MongoDB](/zh-cn/configuration/output-plugins/MongoDB)
     - [MySQL](/zh-cn/configuration/output-plugins/MySQL)
     - [S3](/zh-cn/configuration/output-plugins/S3)
     - [Stdout](/zh-cn/configuration/output-plugins/Stdout)

docs/zh-cn/configuration/input-plugins/Hive.docs

@@ -0,0 +1,11 @@
+@waterdropPlugin
+@pluginGroup input
+@pluginName Hive
+@pluginDesc "从hive读取原始数据"
+@pluginAuthor InterestingLab
+@pluginHomepage https://interestinglab.github.io/waterdrop
+@pluginVersion 1.0.0
+
+@pluginOption
+string pre_sql yes "进行预处理的sql, 如果不需要预处理,可以使用select * from hive_db.hive_table"
+string table_name yes "预处理sql的到数据注册成的临时表名"

docs/zh-cn/configuration/input-plugins/Hive.md

@@ -0,0 +1,40 @@
+## Input plugin : Hive
+
+* Author: InterestingLab
+* Homepage: https://interestinglab.github.io/waterdrop
+* Version: 1.0.0
+
+### Description
+
+从hive中获取数据，
+
+### Options
+
+| name | type | required | default value |
+| --- | --- | --- | --- |
+| [pre_sql](#pre_sql-string) | string | yes | - |
+| [table_name](#table_name-string) | string | yes | - |
+
+
+##### pre_sql [string]
+
+进行预处理的sql, 如果不需要预处理,可以使用select * from hive_db.hive_table
+
+##### table_name [string]
+
+经过pre_sql获取到的数据，注册成临时表的表名
+
+
+
+### Example
+
+```
+hive {
+    pre_sql = "select * from mydb.mytb"
+    table_name = "myTable"
+}
+```
+
+### Notes
+cluster和client模式下必须把hadoopConf和hive-site.xml置于集群每个节点sparkconf目录下,本地调试将其放在resources目录
+

docs/zh-cn/configuration/input-plugins/Kudu.docs

@@ -0,0 +1,12 @@
+@waterdropPlugin
+@pluginGroup input
+@pluginName Kudu
+@pluginDesc "从[Apache Kudu](https://kudu.apache.org) 表中读取数据"
+@pluginAuthor InterestingLab
+@pluginHomepage https://interestinglab.github.io/waterdrop
+@pluginVersion 1.0.0
+
+@pluginOption
+string kudu_master yes "kudu的master，多个master以逗号隔开"
+string kudu_table yes "kudu要读取的表名"
+string table_name yes "获取到数据注册成的临时表名"

docs/zh-cn/configuration/input-plugins/Kudu.md

@@ -0,0 +1,42 @@
+## Input plugin : Kudu
+
+* Author: InterestingLab
+* Homepage: https://interestinglab.github.io/waterdrop
+* Version: 1.0.0
+
+### Description
+
+从[Apache Kudu](https://kudu.apache.org) 表中读取数据.
+
+### Options
+
+| name | type | required | default value |
+| --- | --- | --- | --- |
+| [kudu_master](#kudu_master-string) | string | yes | - |
+| [kudu_table](#kudu_table) | string | yes | - |
+| [table_name](#table_name-string) | string | yes | - |
+
+
+##### kudu_master [string]
+
+kudu的master，多个master以逗号隔开
+
+##### kudu_table [string]
+
+kudu中要读取的表名
+
+##### table_name [string]
+
+获取到的数据，注册成临时表的表名
+
+
+
+### Example
+
+```
+kudu{
+   kudu_master="hadoop01:7051,hadoop02:7051,hadoop03:7051"
+   kudu_table="my_kudu_table"
+   table_name="reg_table"
+ }
+```

docs/zh-cn/configuration/input-plugins/MongoDB.docs

@@ -0,0 +1,13 @@
+@waterdropPlugin
+@pluginGroup input
+@pluginName MongoDB
+@pluginDesc "从[MongoDB](https://www.mongodb.com/)读取数据"
+@pluginAuthor InterestingLab
+@pluginHomepage https://interestinglab.github.io/waterdrop
+@pluginVersion 1.0.0
+
+@pluginOption
+string readConfig.uri yes "mongoDB uri"
+string readConfig.database yes "要读取的database"
+string readConfig.collection yes "要读取的collection"
+string table_name yes "读取数据注册成的临时表名"

docs/zh-cn/configuration/input-plugins/MongoDB.md

@@ -0,0 +1,54 @@
+## Input plugin : MongoDB
+
+* Author: InterestingLab
+* Homepage: https://interestinglab.github.io/waterdrop
+* Version: 1.0.0
+
+### Description
+
+从[MongoDB](https://www.mongodb.com/)读取数据
+
+### Options
+
+| name | type | required | default value |
+| --- | --- | --- | --- |
+| [readconfig.uri](#readconfig.uri-string) | string | yes | - |
+| [readconfig.database](#readconfig.database-string) | string | yes | - |
+| [readconfig.collection](#readconfig.collection-string) | string | yes | - |
+| [readconfig.*](#readconfig.*-string) | string | no | - |
+| [table_name](#table_name-string) | string | yes | - |
+
+
+##### readconfig.uri [string]
+
+要读取mongoDB的uri
+
+##### readconfig.database [string]
+
+要读取mongoDB的database
+
+##### readconfig.collection [string]
+
+要读取mongoDB的collection
+
+#### readconfig
+
+这里还可以配置更多其他参数，详见https://docs.mongodb.com/spark-connector/v1.1/configuration/, 参见其中的`Input Configuration`部分
+指定参数的方式是在原参数名称上加上前缀"readconfig." 如设置`spark.mongodb.input.partitioner`的方式是 `readconfig.spark.mongodb.input.partitioner="MongoPaginateBySizePartitioner"`。如果不指定这些非必须参数，将使用MongoDB官方文档的默认值
+
+##### table_name [string]
+
+从mongoDB获取到的数据，注册成临时表的表名
+
+
+### Example
+
+```
+mongodb{
+        readconfig.uri="mongodb://myhost:mypost"
+        readconfig.database="mydatabase"
+        readconfig.collection="mycollection"
+        readconfig.spark.mongodb.input.partitioner = "MongoPaginateBySizePartitioner"
+        table_name = "test"
+      }
+```

docs/zh-cn/configuration/output-plugins/Kudu.docs

@@ -0,0 +1,12 @@
+@waterdropPlugin
+@pluginGroup output
+@pluginName Kudu
+@pluginDesc "写入数据到[Apache Kudu](https://kudu.apache.org)表中"
+@pluginAuthor InterestingLab
+@pluginHomepage https://interestinglab.github.io/waterdrop
+@pluginVersion 1.0.0
+
+@pluginOption
+string kudu_master yes "kudu的master，多个master以逗号隔开"
+string kudu_table yes "kudu中要写入的表名,表必须已经存在"
+string mode="insert" no "写入kudu模式 insert|update|upsert|insertIgnore"

docs/zh-cn/configuration/output-plugins/Kudu.md

@@ -0,0 +1,43 @@
+## Output plugin : Kudu
+
+* Author: InterestingLab
+* Homepage: https://interestinglab.github.io/waterdrop
+* Version: 1.0.0
+
+### Description
+
+写入数据到[Apache Kudu](https://kudu.apache.org)表中
+
+### Options
+
+| name | type | required | default value |
+| --- | --- | --- | --- |
+| [kudu_master](#kudu_master-string) | string | yes | - |
+| [kudu_table](#kudu_table) | string | yes | - |
+| [mode](#mode-string) | string | no | insert |
+
+
+##### kudu_master [string]
+
+kudu的master，多个master以逗号隔开
+
+##### kudu_table [string]
+
+kudu中要写入的表名,表必须已经存在
+
+##### mode [string]
+
+写入kudu中采取的模式,支持 insert|update|upsert|insertIgnore,默认为insert
+insert和insertIgnore :insert在遇见主键冲突将会报错，insertIgnore不会报错，将会舍弃这条数据
+update和upsert :update找不到要更新的主键将会报错，upsert不会，将会把这条数据插入
+
+
+### Example
+
+```
+kudu{
+   kudu_master="hadoop01:7051,hadoop02:7051,hadoop03:7051"
+   kudu_table="my_kudu_table"
+   mode="upsert"
+ }
+```

docs/zh-cn/configuration/output-plugins/MongoDB.docs

@@ -0,0 +1,12 @@
+@waterdropPlugin
+@pluginGroup output
+@pluginName MongoDB
+@pluginDesc "写入数据到[MongoDB](https://www.mongodb.com/)"
+@pluginAuthor InterestingLab
+@pluginHomepage https://interestinglab.github.io/waterdrop
+@pluginVersion 1.0.0
+
+@pluginOption
+string readConfig.uri yes "mongoDB uri"
+string readConfig.database yes "要写入的database"
+string readConfig.collection yes "要写入的collection"

docs/zh-cn/configuration/output-plugins/MongoDB.md

@@ -0,0 +1,49 @@
+## Output plugin : MongoDB
+
+* Author: InterestingLab
+* Homepage: https://interestinglab.github.io/waterdrop
+* Version: 1.0.0
+
+### Description
+
+写入数据到[MongoDB](https://www.mongodb.com/)
+
+### Options
+
+| name | type | required | default value |
+| --- | --- | --- | --- |
+| [writeconfig.uri](#writeconfig.uri-string) | string | yes | - |
+| [writeconfig.database](#writeconfig.database-string) | string | yes | - |
+| [writeconfig.collection](#writeconfig.collection-string) | string | yes | - |
+| [writeconfig.*](#writeconfig.*-string) | string | no | - |
+
+
+
+##### writeconfig.uri [string]
+
+要写入mongoDB的uri
+
+##### writeconfig.database [string]
+
+要写入mongoDB的database
+
+##### writeconfig.collection [string]
+
+要写入mongoDB的collection
+
+#### writeconfig
+
+这里还可以配置更多其他参数，详见https://docs.mongodb.com/spark-connector/v1.1/configuration/
+, 参见其中的`Output Configuration`部分
+指定参数的方式是在原参数名称上加上前缀"writeconfig." 如设置`localThreshold`的方式是 `writeconfig.localThreshold=20`。如果不指定这些非必须参数，将使用MongoDB官方文档的默认值
+
+
+### Example
+
+```
+mongodb{
+        writeconfig.uri="mongodb://myhost:mypost"
+        writeconfig.database="mydatabase"
+        writeconfig.collection="mycollection"
+      }
+```

waterdrop-core/build.sbt

@@ -33,6 +33,9 @@ libraryDependencies ++= Seq(
     exclude("org.spark-project.spark", "unused")
     exclude("net.jpountz.lz4", "unused"),
   "com.typesafe" % "config" % "1.3.1",
+  "org.apache.spark" %% "spark-hive" % sparkVersion ,
+  "org.mongodb.spark" %% "mongo-spark-connector" % "2.2.0",
+  "org.apache.kudu" %% "kudu-spark2" % "1.7.0",
   "com.alibaba" % "QLExpress" % "3.2.0",
   "com.alibaba" % "fastjson" % "1.2.47",
   "commons-lang" % "commons-lang" % "2.6",
@@ -41,7 +44,7 @@ libraryDependencies ++= Seq(
   "org.elasticsearch" % "elasticsearch-spark-20_2.11" % "5.6.3",
   "com.github.scopt" %% "scopt" % "3.7.0",
   "org.apache.commons" % "commons-compress" % "1.15",
-  "ru.yandex.clickhouse" % "clickhouse-jdbc" % "0.1.39" excludeAll(ExclusionRule(organization="com.fasterxml.jackson.core"))
+  "ru.yandex.clickhouse" % "clickhouse-jdbc" % "0.1.39" exclude("com.google.guava","guava") excludeAll(ExclusionRule(organization="com.fasterxml.jackson.core"))
 )
 
 // For binary compatible conflicts, sbt provides dependency overrides.

waterdrop-core/src/main/resources/META-INF/services/io.github.interestinglab.waterdrop.apis.BaseOutput

@@ -6,4 +6,6 @@ io.github.interestinglab.waterdrop.output.Jdbc
 io.github.interestinglab.waterdrop.output.Kafka
 io.github.interestinglab.waterdrop.output.Mysql
 io.github.interestinglab.waterdrop.output.S3
-io.github.interestinglab.waterdrop.output.Stdout
+io.github.interestinglab.waterdrop.output.Stdout
+io.github.interestinglab.waterdrop.output.MongoDB
+io.github.interestinglab.waterdrop.output.Kudu

waterdrop-core/src/main/resources/META-INF/services/io.github.interestinglab.waterdrop.apis.BaseStaticInput

@@ -1,3 +1,6 @@
 io.github.interestinglab.waterdrop.input.Fake2
 io.github.interestinglab.waterdrop.input.Hdfs
-io.github.interestinglab.waterdrop.input.File
+io.github.interestinglab.waterdrop.input.File
+io.github.interestinglab.waterdrop.input.Hive
+io.github.interestinglab.waterdrop.input.MongoDB
+io.github.interestinglab.waterdrop.input.Kudu

waterdrop-core/src/main/scala/io/github/interestinglab/waterdrop/Waterdrop.scala

@@ -172,7 +172,7 @@ object Waterdrop extends Logging {
       println("\t" + key + " => " + value)
     })
 
-    val sparkSession = SparkSession.builder.config(sparkConf).getOrCreate()
+    val sparkSession = SparkSession.builder.config(sparkConf).enableHiveSupport().getOrCreate()
 
     // find all user defined UDFs and register in application init
     UdfRegister.findAndRegisterUdfs(sparkSession)