Как просмотреть временные таблицы pyspark на сервере Thrift?

Question

Я пытаюсь сделать временную таблицу созданной в pyspark доступной через Thrift. Моя конечная цель - получить доступ к нему из клиента базы данных, такого как DBeaver, с помощью JDBC.

Сначала я тестирую с использованием билайна.

Это то, что я делаю.

1. Запустил кластер с одним рабочим на моей машине, используя docker, и добавил spark.sql.hive.thriftServer.singleSession true на spark-defaults.conf

2. Запустил оболочку Pyspark (для тестирования сакэ) и запустил следующий код:

   from pyspark.sql import Row l = [('Ankit',25),('Jalfaizy',22),('saurabh',20),('Bala',26)] rdd = sc.parallelize(l) people = rdd.map(lambda x: Row(name=x[0], age=int(x[1]))) people = people.toDF().cache() peebs = people.createOrReplaceTempView('peebs') result = sqlContext.sql('select * from peebs')

   Пока все хорошо, все работает отлично.

3. На другом терминале я инициализирую сервер spark thrift: ./sbin/start-thriftserver.sh --hiveconf hive.server2.thrift.port=10001 --conf spark.executor.cores=1 --master spark://172.18.0.2:7077

   Сервер, кажется, запускается нормально, и я могу видеть задания как pyspark, так и thrift-сервера, работающие на главном интерфейсе моего спарк-кластера.

4. потом подключаюсь к кластеру с помощью билайна

   ./bin/beeline beeline> !connect jdbc:hive2://172.18.0.2:10001

   Вот что я получил

   Подключение к jdbc: hive2: //172.18.0.2: 10001
   Введите имя пользователя для jdbc: hive2: //172.18.0.2: 10001:
   Введите пароль для jdbc: hive2: //172.18.0.2: 10001:
   2019-06-29 20:14:25 INFO Utils: 310 - Предоставленные полномочия: 172.18.0.2:10001
   2019-06-29 20:14:25 INFO Utils: 397 - Разрешенные права доступа: 172.18.0.2:10001
   2019-06-29 20:14:25 ИНФОРМАЦИЯ HiveConnection: 203 - Попытка открыть клиентский транспорт с JDBC Uri: jdbc: hive2: //172.18.0.2: 10001
   Подключено к: Spark SQL (версия 2.3.3)
   Драйвер: Hive JDBC (версия 1.2.1.spark2)
   Изоляция транзакции: TRANSACTION_REPEATABLE_READ

   Кажется, все в порядке.

5. Когда я перечисляю show tables; Я ничего не вижу.

Я хотел бы выделить две интересные вещи:

1. Когда я запускаю pyspark, я получаю эти предупреждения

   WARN ObjectStore: 6666 - Информация о версии не найдена в metastore. hive.metastore.schema.verification не включен, поэтому запись схемы версии 1.2.0

   WARN ObjectStore: 568 - Не удалось получить базу данных по умолчанию, возвращая NoSuchObjectException

   WARN ObjectStore: 568 - Не удалось получить базу данных global_temp, возвращая NoSuchObjectException

2. Когда я запускаю комиссионный сервер, я получаю:

   rsync от искры: //172.18.0.2: 7077
   ssh: не удалось разрешить искра имени хоста: имя или служба неизвестны
   rsync: соединение неожиданно закрыто (получено 0 байт) [Получатель]
   Ошибка rsync: необъяснимая ошибка (код 255) в io.c (235) [Receiver = 3.1.2]
   запуск org.apache.spark.sql.hive.thriftserver.HiveThriftServer2, ведение журнала в ...

Я прошел через несколько постов и обсуждений. Я вижу людей, которые говорят, что мы не можем открывать временные таблицы с помощью thrift, если вы не запустите сервер из одного и того же кода. Если это правда, как я могу сделать это в python (pyspark)?

Спасибо

Answer 1

После нескольких тестов я смог придумать простой (без аутентификации) код, который работает для меня.

Важно отметить, что если вы хотите сделать временные таблицы доступными через JDBC, вам нужно запустить Thrift-сервер в той же JVM (то же самое задание на запуск) и убедиться, что код зависает, чтобы приложение продолжало работать в кластере.

После рабочего примера кода, который я создал для справки:

import time
from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession
from py4j.java_gateway import java_import

spark = SparkSession \
    .builder \
    .appName('the_test') \
    .enableHiveSupport()\
    .config('spark.sql.hive.thriftServer.singleSession', True)\
    .config('hive.server2.thrift.port', '10001') \
    .getOrCreate()

sc=spark.sparkContext
sc.setLogLevel('INFO')

java_import(sc._gateway.jvm, "")


from pyspark.sql import Row
l = [('John', 20), ('Heather', 34), ('Sam', 23), ('Danny', 36)]
rdd = sc.parallelize(l)
people = rdd.map(lambda x: Row(name=x[0], age=int(x[1])))
people = people.toDF().cache()
peebs = people.createOrReplaceTempView('peebs')

sc._gateway.jvm.org.apache.spark.sql.hive.thriftserver.HiveThriftServer2.startWithContext(spark._jwrapped)

while True:
    time.sleep(10)

Я просто использовал вышеупомянутый .py в моей искровой отправке, и я смог подключиться через JDBC через beeline и с помощью DBeaver с помощью драйвера Jive для Hive.

Answer 2

createOrReplaceTempView создает таблицу в памяти. Комиссионный сервер Spark должен быть запущен на той же JVM драйвера, где мы создали таблицу в памяти.
В приведенном выше примере драйвер, на котором создается таблица, и драйвер, на котором работает STS (сервер Spark Thrift), различаются.
Два варианта
1. Создайте таблицу, используя createOrReplaceTempView в той же JVM, где запущен STS.
2. Используйте резервное хранилище метастазов и создайте таблицы, используя org.apache.spark.sql.DataFrameWriter#saveAsTable, чтобы таблицы были доступны независимо от JVM (фактически без какого-либо драйвера Spark.

По поводу ошибок:
1. Относится к версии метасторана клиента и сервера.
2. Похоже, какой-то скрипт rsync пытается расшифровать spark:\\ url
Кажется, что и то, и другое не связано с этой проблемой.