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
package mobvista.dmp.datasource.iqiyi
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
/**
* @package: mobvista.dmp.datasource.iqiyi
* @author: wangjf
* @date: 2020/4/29
* @time: 11:52 上午
* @email: jinfeng.wang@mobvista.com
* @phone: 152-1062-7698
*/
class IQiYiLaHuoCK extends CommonSparkJob with Serializable {
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 input = commandLine.getOptionValue("input")
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"IQiYiLaHuoCK.$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 {
val rdd = sc.textFile(input).map(r => {
Row(r)
})
implicit val clickhouseDataSource: ClickHouseDataSource = ClickHouseConnectionFactory.get(host)
val clusterName = Some(cluster): Option[String]
val tdf = spark.emptyDataFrame
// drop table
tdf.dropPartition(database, table, date, clusterName)
val dt = mobvista.dmp.common.MobvistaConstant.sdf1.format(mobvista.dmp.common.MobvistaConstant.sdf2.parse(date))
spark.createDataFrame(rdd, schema).saveToClickHouse(database, table, Seq(dt), Seq("dt"), clusterName, batchSize = 1000000)
} finally {
sc.stop()
spark.stop()
}
0
}
val schema = StructType(Array(
StructField("device_ids", StringType)))
override protected def buildOptions(): Options = {
val options = new Options
options.addOption("input", true, "input")
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
}
}
object IQiYiLaHuoCK {
def main(args: Array[String]): Unit = {
new IQiYiLaHuoCK().run(args)
}
}