连接到Apache Spark中的SQLite

Question

我想在SQLite数据库中的所有表上运行一个自定义函数。函数或多或少是相同的，但取决于单个表的模式。此外，表及其模式只有在运行时才是已知的（使用指定数据库路径的参数调用程序）。

这就是我目前所拥有的：

val conf = new SparkConf().setAppName("MyApp")
val sc = new SparkContext(conf)
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
// somehow bind sqlContext to DB
val allTables = sqlContext.tableNames
for( t <- allTables) {
    val df = sqlContext.table(t)
    val schema = df.columns
    sqlContext.sql("SELECT * FROM " + t + "...").map(x => myFunc(x,schema))
}

到目前为止，我发现的唯一提示需要提前了解表格，而在我的场景中并非如此：

val tableData = 
  sqlContext.read.format("jdbc")
    .options(Map("url" -> "jdbc:sqlite:/path/to/file.db", "dbtable" -> t))
    .load()

我使用的是xerial sqlite jdbc驱动程序。那么，我如何才能只连接到数据库，而不连接到表呢？

编辑：使用Berylium的答案作为开始，我将代码更新为：

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
val metaData = sqlContext.read.format("jdbc")
    .options(Map("url" -> "jdbc:sqlite:/path/to/file.db",
                 "dbtable" -> "(SELECT * FROM sqlite_master) AS t")).load()
val myTableNames = metaData.select("tbl_name").distinct()
for (t <- myTableNames) {
    println(t.toString)
    val tableData = sqlContext.table(t.toString)
    for (record <- tableData.select("*")) {
        println(record)
    }
}

至少我可以在运行时读取表名，这对我来说是一个巨大的进步。但我不会读取表。我试过两种

val tableData = sqlContext.table(t.toString)

和

val tableData = sqlContext.read.format("jdbc")
    .options(Map("url" -> "jdbc:sqlite:/path/to/file.db",
                 "dbtable" -> t.toString)).load()

在循环中，但在这两种情况下，我都会得到一个NullPointerException。虽然我可以打印表名，但似乎无法连接到它们。

最后但同样重要的是，我总是收到一个SQLITE_ERROR: Connection is closed错误。这看起来与本问题中描述的问题相同：SQLITE_ERROR：当通过JDBC从Spark连接到SQLITE数据库时，连接关闭

Answer 1

有两个选项可以尝试

直接使用JDBC

• 在Spark作业中打开一个单独的、简单的JDBC连接
• 从JDBC元数据中获取表名
• 将这些内容输入到您的for理解中

对"dbtable"参数使用SQL查询

可以将查询指定为dbtable参数的值。从语法上讲，这个查询必须"看起来"像一个表，所以它必须封装在一个子查询中。

在该查询中，从数据库中获取元数据：

val df = sqlContext.read.format("jdbc").options(
  Map(
    "url" -> "jdbc:postgresql:xxx",
    "user" -> "x",
    "password" -> "x",
    "dbtable" -> "(select * from pg_tables) as t")).load()

这个例子适用于PostgreSQL，你必须将它适应SQLite。

更新

JDBC驱动程序似乎只支持对一个结果集进行迭代。无论如何，当您使用collect()实现表名列表时，下面的代码段应该可以工作：

val myTableNames = metaData.select("tbl_name").map(_.getString(0)).collect()
for (t <- myTableNames) {
  println(t.toString)
  val tableData = sqlContext.read.format("jdbc")
    .options(
      Map(
        "url" -> "jdbc:sqlite:/x.db",
        "dbtable" -> t)).load()
  tableData.show()
}