Connecting PySpark with Hive tables

[jankogasic]

Hello, I am trying to use pyspark to access kyuubi and Spark. The issue I have is with Hive dialect. The queries that end up on cluster have some weird syntax and they fail because of that.

STEPS TO REPRODUCE:

• sudo apt install -y openjdk-17-jdk
• export JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64
• export PATH="$JAVA_HOME/bin:$PATH"
• pip install pyspark 'pyspark[sql]' 'pyspark[pandas_on_spark]'
• sudo apt install -y krb5-user
• sudo cp krb5.conf /etc/
• kinit -kt janko-gasic.keytab [email protected]

import pyspark
print(pyspark.__file__)
print(pyspark.__version__)

from pyspark.sql import SparkSession

DRIVER_JAR  = "./kyuubi-hive-jdbc-shaded-1.10.2.jar"
DIALECT_JAR = "./kyuubi-extension-spark-jdbc-dialect_2.12-1.10.2.jar"

spark = (
    SparkSession.builder
    .appName("KyuubiJDBC")
    .config("spark.jars", f"{DRIVER_JAR},{DIALECT_JAR}")
    # make sure the driver also sees them
    .config("spark.driver.extraClassPath", f"{DRIVER_JAR}:{DIALECT_JAR}")
    # register the dialect (this is what enables ARRAY/MAP/STRUCT over JDBC)
    .config("spark.sql.extensions", "org.apache.spark.sql.dialect.KyuubiSparkJdbcDialectExtension")
    .getOrCreate()
)

jdbc_url = (
    "jdbc:kyuubi://spark1.lan.bla1.us:2181,spark2.lan.bla1.us:2181,"
    ";serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=kyuubi"
    ";auth=KERBEROS;principal=kyuubi/[email protected];ssl=true"
)

try:
    df = (
        spark.read.format("jdbc")
        .option("driver", "org.apache.kyuubi.jdbc.KyuubiHiveDriver")
        .option("url", jdbc_url)
        .option("query", "select 1") # <--------------------------------------------------- simple query
        .load()
    )
except Exception as e:
    print(e)
finally:
    spark.stop()
    
df.printSchema()
df.show(5)

My error is

Py4JJavaError: An error occurred while calling o123.showString.
: java.lang.NullPointerException: Cannot invoke "org.apache.spark.SparkEnv.rpcEnv()" because the return value of "org.apache.spark.SparkEnv$.get()" is null

pyspark 4.0.1 (aslo tried 3.5 but no luck)
kyuubi 1.10.2

[pan3793]

One obvious issue - you are using Scala 2.12 kyuubi-extension-spark-jdbc-dialect_2.12-1.10.2.jar, while Spark 4.0 only supports Scala 2.13. But I think this won't cause the NPE, the single line error message does not help much. (I'm not familiar with PySpark so may not be able to provide more guidance for diagnosis)

[jankogasic]

I made a stupid mistake! I was calling functions on df after closing spark context!

BTW I made it work with pyspark=3.5

[jankogasic]

I just realized I didn't quite solve this yet. When I execute more complex SELECT query, I get error:

An error occurred while calling o108.load.
: org.apache.spark.SparkSQLException: [UNRECOGNIZED_SQL_TYPE] Unrecognized SQL type - name: array, id: ARRAY.
	at org.apache.spark.sql.errors.QueryExecutionErrors$.unrecognizedSqlTypeError(QueryExecutionErrors.scala:993)
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.getCatalystType(JdbcUtils.scala:228)
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.$anonfun$getSchema$1(JdbcUtils.scala:308)
	at scala.Option.getOrElse(Option.scala:189)
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.getSchema(JdbcUtils.scala:308)
	at org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD$.getQueryOutputSchema(JDBCRDD.scala:71)
	at org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD$.resolveTable(JDBCRDD.scala:58)
	at org.apache.spark.sql.execution.datasources.jdbc.JDBCRelation$.getSchema(JDBCRelation.scala:241)
	at org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:37)
	at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:346)
	at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:229)
	at org.apache.spark.sql.DataFrameReader.$anonfun$load$2(DataFrameReader.scala:211)
	at scala.Option.getOrElse(Option.scala:189)
	at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:211)
	at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:172)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:75)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:52)
	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
	at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
	at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:374)
	at py4j.Gateway.invoke(Gateway.java:282)
	at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
	at py4j.commands.CallCommand.execute(CallCommand.java:79)
	at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)
	at py4j.ClientServerConnection.run(ClientServerConnection.java:106)
	at java.base/java.lang.Thread.run(Thread.java:1583)

Fetching simple data types is working, but fetching ARRAY or STRUCT is failing. It seems like Hive dialect translation is not working.

I'm using Kyuubi 1.10.2, pyspark 3.5. Tried both KyuubiHiveDriver and HiveDriver but no luck.

[jankogasic]

Here are few more exmples:

query which I'm executing:   SHOW DATABASES
...........................................................................................
An error occurred while calling o680.load.
: java.sql.SQLException: org.apache.kyuubi.KyuubiSQLException: org.apache.kyuubi.KyuubiSQLException: Error operating ExecuteStatement: org.apache.spark.sql.AnalysisException: [TABLE_OR_VIEW_NOT_FOUND] The table or view `SHOW` cannot be found. Verify the spelling and correctness of the schema and catalog.
If you did not qualify the name with a schema, verify the current_schema() output, or qualify the name with the correct schema and catalog.
To tolerate the error on drop use DROP VIEW IF EXISTS or DROP TABLE IF EXISTS.; line 1 pos 15;



query which I'm executing:   SHOW DATABASES;
...........................................................................................
An error occurred while calling o640.load.
: java.sql.SQLException: org.apache.kyuubi.KyuubiSQLException: org.apache.kyuubi.KyuubiSQLException: Error operating ExecuteStatement: org.apache.spark.sql.catalyst.parser.ParseException: 
[PARSE_SYNTAX_ERROR] Syntax error at or near ';': extra input ';'.(line 1, pos 29)

== SQL ==
SELECT * FROM (SHOW DATABASES;) SPARK_GEN_SUBQ_120 WHERE 1=0
-----------------------------^^^

[jankogasic]

After digging deeper here is what I found:

• Successfully connected with DBever to Kyuubi and viewing ARRAY column as expected
• Still can't pull ARRAY with pyspark. I have found Hive dialect jar definition here doesn't seem to override ARRAY objects. Does it mean that it is not supported? @bowenliang123 I see that you worked on this feature. Do you possibly know the solution?