通过 JDBC 集成 Spark SQL 和 Apache Drill

Question

我想根据使用 Apache Drill 对 CSV 数据（在 HDFS 上）执行的查询结果创建一个 Spark SQL DataFrame。我成功配置了 Spark SQL，让它通过 JDBC 连接到 Drill：

Map<String, String> connectionOptions = new HashMap<String, String>();
connectionOptions.put("url", args[0]);
connectionOptions.put("dbtable", args[1]);
connectionOptions.put("driver", "org.apache.drill.jdbc.Driver");

DataFrame logs = sqlc.read().format("jdbc").options(connectionOptions).load();

Spark SQL 执行两个查询：第一个用于获取架构，第二个用于检索实际数据：

SELECT * FROM (SELECT * FROM dfs.output.`my_view`) WHERE 1=0

SELECT "field1","field2","field3" FROM (SELECT * FROM dfs.output.`my_view`)

第一个是成功的，但是在第二个中，Spark 将字段括在双引号中，这是 Drill 不支持的，因此查询失败。

有人设法使这种集成正常工作吗？

谢谢！

Answer 1

这是接受的答案代码在 Java 中的外观：

import org.apache.spark.sql.jdbc.JdbcDialect;

public class DrillDialect extends JdbcDialect {
  @Override
  public String quoteIdentifier(String colName){
    return colName;
  }

  public boolean canHandle(String url){
    return url.startsWith("jdbc:drill:");
  }
}

在创建 Spark 会话之前注册方言：

import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.jdbc.JdbcDialects;

public static void main(String[] args) {
    JdbcDialects.registerDialect(new DrillDialect());
    SparkSession spark = SparkSession
      .builder()
      .appName("Drill Dialect")
      .getOrCreate();

     //More Spark code here..

    spark.stop();
}

使用 Spark 2.3.2 和 Drill 1.16.0 进行了尝试和测试。希望对你也有帮助！

Answer 2

您可以为此添加 JDBC 方言并在使用 jdbc 连接器之前注册方言

case object DrillDialect extends JdbcDialect {

  def canHandle(url: String): Boolean = url.startsWith("jdbc:drill:")

  override def quoteIdentifier(colName: java.lang.String): java.lang.String = {
    return colName
  }

  def instance = this
}

JdbcDialects.registerDialect(DrillDialect)