Ошибка sqlContext HiveDriver в SQLException: метод не поддерживается

Я пытаюсь использовать sqlContext.read.format("jdbc").options(driver="org.apache.hive.jdbc.HiveDriver") чтобы получить таблицу Hive в Spark без каких-либо успехов. Я провел исследование и прочитал ниже:

Я использовал последнюю версию Hortonworks Sandbox 2.6 и спросил у сообщества тот же вопрос:

https://community.hortonworks.com/info/156828/pyspark-jdbc-py4jjavaerror-calling-o95load-javasql.html?childToView=156936#answer-156936

То, что я хочу сделать, очень просто с помощью pyspark:

df = sqlContext.read.format("jdbc").options(driver="org.apache.hive.jdbc.HiveDriver", url="jdbc:hive2://localhost:10016/default", dbtable="sample_07",user="maria_dev", password="maria_dev").load()

Это дало мне эту ошибку:

17/12/30 19:55:14 INFO HiveConnection: Will try to open client transport with JDBC Uri: jdbc:hive2://localhost:10016/default
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/hdp/current/spark-client/python/pyspark/sql/readwriter.py", line 139, in load
    return self._df(self._jreader.load())
  File "/usr/hdp/current/spark-client/python/lib/py4j-0.9-src.zip/py4j/java_gateway.py", line 813, in __call__
  File "/usr/hdp/current/spark-client/python/pyspark/sql/utils.py", line 45, in deco
    return f(*a, **kw)
  File "/usr/hdp/current/spark-client/python/lib/py4j-0.9-src.zip/py4j/protocol.py", line 308, in get_return_value
py4j.protocol.Py4JJavaError: An error occurred while calling o119.load.
: java.sql.SQLException: Method not supported
at org.apache.hive.jdbc.HiveResultSetMetaData.isSigned(HiveResultSetMetaData.java:143)
at org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD$.resolveTable(JDBCRDD.scala:136)
at org.apache.spark.sql.execution.datasources.jdbc.JDBCRelation.<init>(JDBCRelation.scala:91)
at org.apache.spark.sql.execution.datasources.jdbc.DefaultSource.createRelation(DefaultSource.scala:57)
at org.apache.spark.sql.execution.datasources.ResolvedDataSource$.apply(ResolvedDataSource.scala:158)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:119)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:381)
at py4j.Gateway.invoke(Gateway.java:259)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:209)
at java.lang.Thread.run(Thread.java:748)

Используя beeline, он отлично работает

beeline> !connect jdbc:hive2://localhost:10016/default maria_dev maria_dev
Connecting to jdbc:hive2://localhost:10016/default
Connected to: Spark SQL (version 2.1.1.2.6.1.0-129)
Driver: Hive JDBC (version 1.2.1000.2.6.1.0-129)
Transaction isolation: TRANSACTION_REPEATABLE_READ
0: jdbc:hive2://localhost:10016/default> select * from sample_07 limit 2;
+----------+-------------------------+------------+---------+--+
|   code   |       description       | total_emp  | salary  |
+----------+-------------------------+------------+---------+--+
| 00-0000  | All Occupations         | 134354250  | 40690   |
| 11-0000  | Management occupations  | 6003930    | 96150   |
+----------+-------------------------+------------+---------+--+

Я мог бы также сделать это:

spark = SparkSession.Builder().appName("testapp").enableHiveSupport().‌getOrCreate()
spark.sql("select * from default.sample_07").collect()

Но это прямое чтение метаданных в Hive. Я хотел бы использовать JDBC для Spark Thrift Server для тонкой защиты.

Я мог бы делать PostgreSQL следующим образом:

sqlContext.read.format("jdbc").options(driver="org.postgresql.Driver")

Я также могу использовать Scala java.sql.{DriverManager, Connection, Statement, ResultSet} чтобы создать соединение JDBC как клиентскую сторону, чтобы добраться до Spark. Но это в основном помещает все данные в память и затем повторно создает Dataframe вручную.

Поэтому возникает вопрос: существует ли способ создать Spark dataframe с данными таблицы Hive без загрузки данных в память в клиент JDBC, например Scala, и не использовать SparkSession.Builder() как в примерах выше? Мой вариант использования - то, что мне нужно иметь дело с мелкозернистой безопасностью.

Ответ 1

Я не уверен, правильно ли я понимаю ваш вопрос или нет, но из того, что я понимаю, вам нужно будет получить таблицу улья в фрейм данных, для чего вам не нужно иметь соединение JDBC, в ваших примерах ссылки они пытаясь подключиться к различным базам данных (РСУБД), а не к улью.

Пожалуйста, ознакомьтесь с нижеприведенным подходом, используя контекст hive, который вы можете получить в таблице данных.

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, SQLContext}

def main(args: Array[String]): Unit = {

val sparkConf = new SparkConf().setAppName("APPName")
    val sc = new SparkContext(sparkConf)
    val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc)
    val sqlContext = new SQLContext(sc)

val hive_df = hiveContext.sql("select * from schema.table").first()

//other way
// val hive_df= hiveContext.table ("SchemaName.TableName")

//Below will print the first line
df.first()
//count on dataframe
df.count()

}

Если вы действительно хотите использовать соединение JDBC, у меня есть приведенный ниже пример, который я использовал для базы данных Oracle, что могло бы вам помочь.

val oracle_data = sqlContext.load("jdbc", Map("url" -> "jdbc:oracle:thin:username/password//hostname:2134/databaseName", "dbtable" -> "Your query tmp", "driver" -> "oracle.jdbc.driver.OracleDriver"));

Ответ 2

На самом деле я посмотрел на это. Hotornworks и cloudera прекращают поддержку для подключения к улью из Spark через Thrift Server.

Итак, вы работаете над чем-то, что невозможно.

https://www.cloudera.com/documentation/spark2/latest/topics/spark2_known_issues.html#ki_thrift_server.

Ссылки говорят, что экономия отключена, но это специально, чтобы улей от искры. Я могу подключиться ко всем типам баз данных от искры, кроме улья.

Таким образом, вы должны работать на другой стиль авторизации.

Поскольку искровой объект напрямую соединяется с ульем, они снимают опору для экономии.

Из вашего предыдущего вопроса, он может читать данные, но читает неправильные данные. Spark 2.2 Ошибка экономичного сервера на фрейме данных NumberFormatException при запросе таблицы Hive

Код

>>> df = sqlContext.read.format("jdbc").options(driver="org.apache.hive.jdbc.HiveDriver", url="jdbc:hive2://localhost:10016/default", dbtable="test4",user="hive", password="hive").option("fetchsize", "10").load()
>>> df.select("*").show()
+---+----+
| id|desc|
+---+----+
| id|desc|
| id|desc|
+---+----+

Проблема здесь в улье

По умолчанию в диалекте по умолчанию заключать в кавычки идентификаторы используют двойные кавычки. SQL-запрос, такой как SELECT "dw_date" FROM table… будет проанализирован Hive для выбора строкового литерала вместо столбца с именем "dw_date". Заменив кавычки на обратные, похоже, проблема решена. Однако в моем тесте имена столбцов, полученные из Hive, начинаются с префикса с именем таблицы, например table.dw_date. Но вы не можете напрямую оборачивать галочки вокруг него, как table.dw_date. Кроме того, нам нужно обернуть каждую часть в отдельности

код

import org.apache.spark.sql.jdbc.JdbcDialect
    private case object HiveDialect extends JdbcDialect {
      override def canHandle(url : String): Boolean = url.startsWith("jdbc:hive2")
      override def quoteIdentifier(colName: String): String = {
        colName.split(‘.).map(part => s"'$part'").mkString(".")
      }
    }

Пожалуйста, следуйте посту ниже, чтобы реализовать решение.

https://medium.com/@viirya/custom-jdbc-dialect-for-hive-5dbb694cc2bd

https://medium.com/@huaxing/customize-spark-jdbc-data-source-to-work-with-your-dedicated-database-dialect-beec6519af27

Зарегистрировать диалект

JdbcDialects.registerDialect(HiveDialect)

Тогда улей jdbc работает.