Question

使用产品: iobjects for spark: 10.1.0
OS: windows 10
spark version: 2.3.2
问题详细描述: 使用PostGISFeatureRDDProvider读取postgis数据源失败:

// 读取数据源
    val provider = new PostGISFeatureRDDProvider()
    val sourceRDD = provider.rdd(new Configuration(), sc, readParam, new Query(tableName))

    Exception in thread "main" java.lang.NullPointerException: DataStore is null
  at com.supermap.bdt.rddprovider.jdbc.PostGISFeatureRDDProvider.rdd(PostGISFeatureRDDProvider.scala:49)

此外, provider的param设置与编程指南中的一致, 数据库连接信息确认无误.
期待早日收到您的解答.

Answer 1

您好，使用FeatureRDDProviderFactory.rdd方法能否正常读取PostGIS数据源呢？

Comments

用FeatureRDDProviderFactory.rdd读取错误:

Exception in thread "main" java.lang.RuntimeException: Could not find a SpatialRDDProvider to process the params.
	at com.supermap.bdt.FeatureRDDProviderFactory$$anonfun$createProvider$2.apply(FeatureRDDProviderFactory.scala:19)
	at com.supermap.bdt.FeatureRDDProviderFactory$$anonfun$createProvider$2.apply(FeatureRDDProviderFactory.scala:19)
	at scala.Option.getOrElse(Option.scala:121)
	at com.supermap.bdt.FeatureRDDProviderFactory$.createProvider(FeatureRDDProviderFactory.scala:19)
	at com.supermap.bdt.FeatureRDDProviderFactory$.apply(FeatureRDDProviderFactory.scala:14)

---

你代码是怎么写的，麻烦截图看下

---

附上param:

    params.put("dbtype", "postgis")
    params.put("host", "host")
    params.put("port", "5432")
    params.put("schema", "public")
    params.put("database", "result")
    params.put("user", "root")
    params.put("passwd", "password") 
    params.put("numPartitions", 4) 
    params.put("providerType", "JDBC")

---

测试代码:

val ss = CommonUtils.getSparkSession(this.getClass, "GroupByKey")
    val sc = ss.sparkContext

    // 解析参数
    val paramMap = CommonUtils.json2map(args(0))

    val readParam = ReadData.getReadParams(paramMap("inputPath_type"), paramMap("inputPath"))
    val tableName = paramMap("tableName")
    val output = paramMap("output")
    val aggregationField = paramMap("aggregationField")

    // 读取数据源
    val sourceRDD = FeatureRDDProviderFactory(readParam).rdd(new Configuration(), sc, readParam, new Query(tableName))

    // 计算聚合结果
    val resultRDD = sourceRDD.map(data => (aggregationField, data.getAttribute("SHAPE_Area").toString.toDouble)).reduceByKey(_ + _)

    // 输出结果
    import ss.implicits._
    CommonUtils.writeForPostgres(resultRDD.toDF(aggregationField, "total_result"), output)

    CommonUtils.stopSpark(ss)

---

设置的参数里面host的值直接设置计算机的ip再获取试试

---

没有改善, 还是一样的错误

---

如果您本地测试可以成功读取的话，麻烦把读取的param给我发一个

---

您那边试下这段代码：
   val params = new util.HashMap[String, java.io.Serializable]()
    params.put("dbtype", "postgis")
    params.put("host", "localhost")
    params.put("port", "5432")
    params.put("schema", "public")
    params.put("database", "pggis")
    params.put("user", "postgres")
    params.put("passwd", "123")
    params.put("providerType", "JDBC")
    val query=new Query("teltracktrip2")
    val rdd=FeatureRDDProviderFactory(params).rdd(new Configuration(),sc,params,query)