导入Spark的数据至Hologres

./bin/spark-shell --jars /path/to/postgresql-42.2.18.jar

• Batch方式
  在Spark中准备数据并配置批量导入数据至Hologres的信息。语句如下。

  //准备数据。
  import org.apache.spark.sql.types._
  import org.apache.spark.sql.Row
  
  val data = Seq(
      Row(1L, "test"),
      Row(2L, "test2"),
      Row(3L, "test2"),
      Row(4L, "test2"),
      Row(5L, "test2"),
      Row(6L, "test2")
  )
  val schema = StructType(Array(
   StructField("a", LongType),
   StructField("b", StringType)
  ))
  
  val df = spark.createDataFrame(spark.sparkContext.parallelize(data),schema)
  
  // 配置导入数据的信息。
  .option("checkpointLocation", checkpointLocation)
  df.write.format("jdbc")
    .option("url", "jdbc:postgresql://endpoint:port/database")
    .option("driver", "org.postgresql.Driver")
    .option("dbtable", "table")//Hologres用于接收数据的表名称。
    .option("user", "accesskey id")//当前阿里云账号的AccessKey ID。
    .option("password", "accesskey secret")//当前阿里云账号的Accesskey Secret。
    .option("isolationLevel", "NONE")
    .save()

  参数说明如下表所示。

  参数	描述
  jdbc:postgresql://endpoint:port/database	endpoint：Hologres实例的网络地址。 port：Hologres实例的端口。 database：Hologres的数据库名称。 您可以进入Hologres管理控制台的实例详情页，从实例配置获取实例地址和端口，从DB管理获取数据库名称。
  table	Hologres用于接收数据的表名称。
  accesskey id	当前阿里云账号的AccessKey ID。 您可以单击用户信息管理，获取AccessKey ID。
  accesskey secret	当前阿里云账号的AccessKey Secret。 您可以单击用户信息管理，获取AccessKey Secret。

• Stream方式
  1. 生成JDBC JAR文件。
     开源Spark或EMR Spark均可以使用如下命令获取JDBC Driver，并编译生成emr-jdbc_2.11-2.1.0-SNAPSHOT.jar文件。

     ./bin/spark-shell --jars /path_to/emr-jdbc_2.11-2.1.0-SNAPSHOT.jar,postgresql-42.2.18.jar --driver-class-path /path_to/emr-jdbc_2.11-2.1.0-SNAPSHOT.jar,postgresql-42.2.18.jar

  2. 在Spark Shell中执行如下命令使用JDBC JAR文件。

     ./bin/spark-shell --jars /path_to/emr-jdbc_2.11-2.1.0-SNAPSHOT.jar,postgresql-42.2.18.jar --driver-class-path /path_to/emr-jdbc_2.11-2.1.0-SNAPSHOT.jar,postgresql-42.2.18.jar

  3. 准备Spark数据并导入至Hologres。
     在Spark中创建一张表并写入数据。配置导入数据至Hologres的信息，并使用流方式实时导入数据至Hologres。

     val wordCounts = lines.flatMap(_.split(" ")).groupBy("value").count()
     
     val query = wordCounts.writeStream
       .outputMode("complete")
       .format("jdbc2")
       .option("url", "jdbc:postgresql://endpoint:port/database")//配置为相应Hologres的实例信息。
       .option("driver", "org.postgresql.Driver")
       .option("dbtable", "table")//Hologres用于接收数据的表名称。
       .option("user", "accesskey id")//当前阿里云账号的AccessKey ID。
       .option("password", "accesskey secret")//当前阿里云账号的AccessKey Secret。
       .option("batchsize", 100) //配置批量导入数据的条数，此处示例配置为100条。
       .option("isolationLevel", "NONE")
       .option("checkpointLocation", checkpointLocation)
       .start()
     
     query.awaitTermination()

     说明

     • 合理设置batchsize参数可以提高导入数据的性能。
     • 当Provider的名称配置为jdbc2时，流计算模式的EMR-SDK性能更好。开源Spark和EMR Spark均可以直接使用jdbc2。