package mobvista.dmp.datasource.behavior import java.net.URI import mobvista.dmp.common.CommonSparkJob import org.apache.commons.cli.Options import org.apache.hadoop.fs.{FileSystem, Path} import org.apache.spark.sql.{Row, SaveMode, SparkSession} class ThirdPartySourceTotal extends CommonSparkJob { override protected def buildOptions(): Options = { val options = new Options options.addOption("outputtotal", true, "[must] outputtotal") options.addOption("dmpevent",true, "[must] dmpevent") options.addOption("coalesce", true, "[must] coalesce") options.addOption("yesterday", true, "[must] yesterday") options } 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 outputtotal = commandLine.getOptionValue("outputtotal") val dmpevent = commandLine.getOptionValue("dmpevent") val coalesce = commandLine.getOptionValue("coalesce") val yesterday = commandLine.getOptionValue("yesterday") val spark = SparkSession.builder() .appName("Behavior_thirdparty_Total") .config("spark.rdd.compress", "true") .config("spark.io.compression.codec", "snappy") .config("spark.sql.orc.filterPushdown", "true") .config("spark.sql.warehouse.dir", "s3://mob-emr-test/spark-warehouse") .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") .enableHiveSupport() .getOrCreate() FileSystem.get(new URI(s"s3://mob-emr-test"), spark.sparkContext.hadoopConfiguration).delete(new Path(outputtotal), true) FileSystem.get(new URI(s"s3://mob-emr-test"), spark.sparkContext.hadoopConfiguration).delete(new Path(dmpevent), true) try { val sql1 = s""" SELECT device_id,country,device_type,platform,tag_name,tag_value,package_name,event_day |from dwh.etl_behavior_thirdparty_data_total where dt ='${yesterday}' """.stripMargin spark.sql(sql1).coalesce(coalesce.toInt) .write .mode(SaveMode.Overwrite) .option("orc.compress", "zlib") .orc(outputtotal) spark.sql(sql1).coalesce(coalesce.toInt) .write .mode(SaveMode.Overwrite) .option("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.SnappyCodec") .parquet(dmpevent) } finally { spark.stop() } 0 } } object ThirdPartySourceTotal { def main(args: Array[String]): Unit = { new ThirdPartySourceTotal().run(args) } }