Spark：导入数据到oracle

方案一：

    //overwrite JdbcDialect fitting for Oracle
    val OracleDialect = new JdbcDialect {
      override def canHandle(url: String): Boolean = url.startsWith("jdbc:oracle") || url.contains("oracle")

      //getJDBCType is used when writing to a JDBC table
      override def getJDBCType(dt: DataType): Option[JdbcType] = dt match {
        case StringType => Some(JdbcType("VARCHAR2(255)", java.sql.Types.VARCHAR))
        case BooleanType => Some(JdbcType("NUMBER(1)", java.sql.Types.NUMERIC))
        case IntegerType => Some(JdbcType("NUMBER(16)", java.sql.Types.NUMERIC))
        case LongType => Some(JdbcType("NUMBER(16)", java.sql.Types.NUMERIC))
        case DoubleType => Some(JdbcType("NUMBER(16,4)", java.sql.Types.NUMERIC))
        case FloatType => Some(JdbcType("NUMBER(16,4)", java.sql.Types.NUMERIC))
        case ShortType => Some(JdbcType("NUMBER(5)", java.sql.Types.NUMERIC))
        case ByteType => Some(JdbcType("NUMBER(3)", java.sql.Types.NUMERIC))
        case BinaryType => Some(JdbcType("BLOB", java.sql.Types.BLOB))
        case TimestampType => Some(JdbcType("DATE", java.sql.Types.DATE))
        case DateType => Some(JdbcType("DATE", java.sql.Types.DATE))
        //        case DecimalType.Fixed(precision, scale) => Some(JdbcType("NUMBER(" + precision + "," + scale + ")", java.sql.Types.NUMERIC))
        case DecimalType.Unlimited => Some(JdbcType("NUMBER(38,2)", java.sql.Types.NUMERIC))
        case _ => None
      }
    }
    //Registering the OracleDialect
    JdbcDialects.registerDialect(OracleDialect)

    val connectProperties = new java.util.Properties()
    connectProperties.put("user", username)
    connectProperties.put("password", password)
    Class.forName(driver).newInstance()

    //write back Oracle
    //Note: When writing the results back orale, be sure that the target table existing
    JdbcUtils.saveTable(mr_case_df, oracleDriverUrl, "MR", connectProperties)

方案二：

val test_df=hiveContext.sql("select * from test")

test_df.foreachPartition(rows => {
  Class.forName(driver)

  val connection: Connection = DriverManager.getConnection(oracleDriverUrl, username, password)
  val prepareStatement: PreparedStatement = connection.prepareStatement("insert into RES_CELL(City,Latiude,longitude)values(?,?,?);")

  rows.foreach(row => {
    prepareStatement.setString(1, row.getAs[String]("city"))      
    prepareStatement.setString(2, row.getAs[String]("latitude"))
    prepareStatement.setString(3, row.getAs[String]("longitude"))
    prepareStatement.addBatch()
  })
  prepareStatement.executeBatch()

  prepareStatement.close()
  connection.close()
})

方案三：

使用sqlloader从spark任务提交节点读取文件导入到oracle。

为什么操作，原因直接从spark中读取hive中的数据使用网络IO连接到集群外的oracle服务器是spark集群不乐意做的事情，对SPARK宝贵的网络IO来说开销特别大。