val prop = new Properties()
prop.setProperty("user", "xxx")
prop.setProperty("password", "xxx")
val url = "jdbc:mysql://ip:port/cdp?useUnicode=true&characterEncoding=utf-8"
spark.read.
option("driver", "com.mysql.jdbc.Driver").
jdbc(url, "client_group", prop).
// jdbc(url, "client_group", Array("proc_status=1", "opt_type!=2", "expired_time < now()"), prop).
persist(StorageLevel.MEMORY_ONLY).
createOrReplaceTempView("tv_client_group")
以下方法并行处理,但要注意条件是否包含重合数据
jdbc(url, "client_group", Array("proc_status=1", "opt_type!=2", "expired_time < now()"), prop)
方法2
val dataframe = spark.read.
format("jdbc").
option("driver", "com.mysql.jdbc.Driver").
option("dbtable", "dbTableName").
option("url", url).
option("user", "userName").
option("password", "password").
load()
写
调用write方法
oraclesql(
s"""SELECT
| cast(MEMBER_ID as INT) MEMBER_ID,
| IF(sale_channel = '1', CAST(NULL AS STRING), store_id) STORE_CODE,
| score INDI_VALUE,
| 1 IS_LATEST,
| CAST('$quarterDate' AS TIMESTAMP) CREATE_TIME,
| bu_id BU_CODE,
| CAST(NULL AS INT) MEMBER_ID_TMP
|FROM recommendation.member_attribute_store
|WHERe create_date='$quarterDate'
|""".stripMargin).
write.option("driver", "oracle.jdbc.driver.OracleDriver").
mode(SaveMode.Append).jdbc(url, tableName, prop)
