1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
package mobvista.dmp.datasource.taobao
import mobvista.dmp.common.CommonSparkJob
import mobvista.dmp.util.PropertyUtil
import mobvista.dmp.utils.clickhouse.ClickHouseConnectionFactory
import mobvista.dmp.utils.clickhouse.ClickHouseSparkExt._
import org.apache.commons.cli.Options
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SparkSession}
import ru.yandex.clickhouse.ClickHouseDataSource
class YOUKUImeiLaXinCK extends CommonSparkJob with Serializable {
override protected def buildOptions(): Options = {
val options = new Options
options.addOption("imeiInput", true, "imeiInput")
options.addOption("date", true, "date")
options.addOption("host", true, "host")
options.addOption("cluster", true, "cluster")
options.addOption("database", true, "database")
options.addOption("table", true, "table")
options
}
val schema = StructType(Array(
StructField("device_ids", StringType)))
override protected def run(args: Array[String]): Int = {
val commandLine = commParser.parse(options, args)
if (!checkMustOption(commandLine)) {
printUsage(options)
return -1
} else printOptions(commandLine)
val imeiInput = commandLine.getOptionValue("imeiInput")
val date = commandLine.getOptionValue("date")
val cluster = commandLine.getOptionValue("cluster")
val host = commandLine.getOptionValue("host")
val database = commandLine.getOptionValue("database")
val table = commandLine.getOptionValue("table")
val spark = SparkSession
.builder()
.appName(s"YOUKUImeiLaXinCK.$date")
.config("spark.rdd.compress", "true")
.config("spark.io.compression.codec", "lz4")
.config("spark.sql.warehouse.dir", "s3://mob-emr-test/spark-warehouse")
.config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.config("spark.clickhouse.driver", "ru.yandex.clickhouse.ClickHouseDriver")
.config("spark.clickhouse.url", PropertyUtil.getProperty("config.properties", "spark.clickhouse.url"))
.config("spark.clickhouse.connection.per.executor.max", "5")
.config("spark.clickhouse.metrics.enable", "true")
.config("spark.clickhouse.socket.timeout.ms", "300000")
.config("spark.clickhouse.cluster.auto-discovery", "true")
.enableHiveSupport()
.getOrCreate()
val sc = spark.sparkContext
try {
try {
//存储imeimd5数据到clickhouse
val imeiRDD = sc.textFile(imeiInput).map(r => {
Row(r)
})
implicit val clickhouseDataSource: ClickHouseDataSource = ClickHouseConnectionFactory.get(host)
val clusterName = Some(cluster): Option[String]
//存储imei到clickhouse
val device_type="IMEI_MD5"
val imeiDF = spark.emptyDataFrame
// drop table
imeiDF.dropPartition(database, table, s"($date,'$device_type')", clusterName)
val dt = mobvista.dmp.common.MobvistaConstant.sdf1.format(mobvista.dmp.common.MobvistaConstant.sdf2.parse(date))
spark.createDataFrame(imeiRDD, schema).saveToClickHouse(database, table, Seq(dt,device_type), Seq("dt","device_type"), clusterName, batchSize = 1000000)
} finally {
sc.stop()
spark.stop()
}
0
}
}
}
object YOUKUImeiLaXinCK {
def main(args: Array[String]): Unit = {
new YOUKUImeiLaXinCK().run(args)
}
}