数据库连接并获取数据：
JavaSparkContext sparkContext = new JavaSparkContext(new SparkConf().setAppName("sparkApp").setMaster("local[5]"));
SQLContext SQLContext = new SQLContext(sparkContext);
Properties Properties = new Properties();
Properties.put("user", "数据库用户名");
Properties.put("password", "数据库密码");
Properties.put("driver", "数据库驱动");
DataFrame DF = SQLContext.read().jdbc("数据库地址","表名",Properties).;
DF：从数据库中获取的数据
数据库写入数据：
/**
* 1、创建类型为Row的RDD
*/
JavaRDD> logDateRdd = sparkContext.parallelize(logDate);
JavaRDD RDD = logDateRdd.map(new Function,Row>(){
@Override
public Row call(List logDate) throws Exception {
return RowFactory.create(
logDate.get(0),
logDate.get(1)
);
}
});
/**
* 2、动态构造DataFrame的元数据。
*/
List structFields = new ArrayList();
structFields.add(DataTypes.createStructField("col1",DataTypes.StringType,false));
structFields.add(DataTypes.createStructField("col2",DataTypes.StringType,true));
//构建StructType，用于最后DataFrame元数据的描述
StructType structType = DataTypes.createStructType(structFields);
/**
* 3、基于已有的元数据以及RDD来构造DataFrame
*/
DataFrame DF = SQLContext.createDataFrame(RDD,structType);
/**
* 4、将数据写入到e_trade_acct_data表中
*/
DF.write().mode("append").jdbc("数据库地址","表名","存有用户名、密码、驱动的Properties类");
sparkContext.parallelize(logDate)：将数据转成RDD
structFields：里面的col1 、col2为数据库字段名，DateTypes 表示数据类型，数据类型要保持一致。false:表示不能为null .true表示可为null