数据库连接并获取数据:
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