|
package com.xiaomi.miui.ad.sparksql.JDBC import org.apache.spark.sql._ /** * Created by michael on 5/23/17. */ object JDBCDataSource { def main(args: Array[String]): Unit = { /* if (args.length < 9) { System.err.println("Usage: JDBCDataSource 2<input> <output>") System.exit(1) }*/ // val Array(input, output) = args val spark = SparkSession .builder() .appName("Spark Codelab: ad_event_join in scala") .getOrCreate() val reader: DataFrameReader = spark.sqlContext.read.format("jdbc") reader.option("url", "jdbc:mysql://主機:端口/數(shù)據(jù)庫") reader.option("dbtable", "表1") reader.option("driver", "com.mysql.jdbc.Driver") reader.option("user", "用戶名") reader.option("password", "密碼") val tagDF: DataFrame = reader.load() reader.option("dbtable", "表2") val mediaDF: DataFrame = reader.load() tagDF.createOrReplaceTempView("dim_tag_id") mediaDF.createOrReplaceTempView("dim_media_type") val sql = "select tag_id,b.media_type,a.type from dim_media_type a join dim_tag_id b " + "on a.media_type = b.media_type" val sql2 = "insert into test_dim_tag_id(tag_id,media_type,display_name,ad_form_id,status," + "input_mode,material_level,deliver_type,creater) select tag_id,media_type,display_name," + "ad_form_id,status,input_mode,material_level,deliver_type,creater from dim_tag_id" val sql3 = "select tag_id,media_type,display_name,ad_form_id,status,input_mode,material_level,deliver_type,creater from dim_tag_id" val sql11: DataFrame = spark.sqlContext.sql(sql) // val sql12: DataFrame = spark.sqlContext.sql(sql2) val prop = new java.util.Properties prop.setProperty("user","用戶名") prop.setProperty("password","密碼") prop.setProperty("driver","com.mysql.jdbc.Driver") val dataResult = spark.sqlContext.sql(sql3).write.mode(SaveMode.Append).format("jdbc") .jdbc("jdbc:mysql://主機:端口/數(shù)據(jù)庫?useUnicode=true&&characterEncoding=utf8","test_dim_tag_id",prop) //插入數(shù)據(jù)庫 } } |
|
|
