摘要：已更新至，归管了，因此也相应统一。本文不再适用及以上版本。字段类型会非常非常奇葩。。。。但是如果体积过于庞大，很容易导致特别是我们一般不会给配置过高的内存。第二个，是函数的返回值。对于而言，我们可以直接使用，来得到这个什么都没有的东西。
Spark已更新至2.x，DataFrame归DataSet管了，因此API也相应统一。本文不再适用2.0.0及以上版本。
DataFrame原生支持直接输出到JDBC，但如果目标表有自增字段（比如id），那么DataFrame就不能直接进行写入了。因为DataFrame.write().jdbc()要求DataFrame的schema与目标表的表结构必须完全一致（甚至字段顺序都要一致），否则会抛异常，当然，如果你SaveMode选择了Overwrite，那么Spark删除你原有的表，然后根据DataFrame的Schema生成一个。。。。字段类型会非常非常奇葩。。。。
于是我们只能通过DataFrame.collect()，把整个DataFrame转成List
翻看Spark的JDBC源码，发现实际上是通过foreachPartition方法，在DataFrame每一个分区中，对每个Row的数据进行JDBC插入，那么为什么我们就不能直接用呢？
Spark JdbcUtils.scala部分源码：
def saveTable(df: DataFrame,url: String,table: String,properties: Properties = new Properties()) {
val dialect = JdbcDialects.get(url)
val nullTypes: Array[Int] = df.schema.fields.map { field =>
dialect.getJDBCType(field.dataType).map(_.jdbcNullType).getOrElse(
field.dataType match {
case IntegerType => java.sql.Types.INTEGER
case LongType => java.sql.Types.BIGINT
case DoubleType => java.sql.Types.DOUBLE
case FloatType => java.sql.Types.REAL
case ShortType => java.sql.Types.INTEGER
case ByteType => java.sql.Types.INTEGER
case BooleanType => java.sql.Types.BIT
case StringType => java.sql.Types.CLOB
case BinaryType => java.sql.Types.BLOB
case TimestampType => java.sql.Types.TIMESTAMP
case DateType => java.sql.Types.DATE
case t: DecimalType => java.sql.Types.DECIMAL
case _ => throw new IllegalArgumentException(
s"Can"t translate null value for field $field")
})
}
val rddSchema = df.schema
val driver: String = DriverRegistry.getDriverClassName(url)
val getConnection: () => Connection = JDBCRDD.getConnector(driver, url, properties)
// ****************** here ******************
df.foreachPartition { iterator =>
savePartition(getConnection, table, iterator, rddSchema, nullTypes)
}
}
嗯。。。既然Scala能实现，那么作为他的爸爸，Java也应该能玩！
我们看看foreachPartition的方法原型：
def foreachPartition(f: Iterator[Row] => Unit)
又是函数式语言最爱的匿名函数。。。非常讨厌写lambda，所以我们还是实现个匿名类吧。要实现的抽象类为：
scala.runtime.AbstractFunction1
来玩耍一下吧！
