Как связать искру с ульем с помощью pyspark? - PullRequest
Новая
       30

Как связать искру с ульем с помощью pyspark?

0 голосов
/ 25 марта 2019

Я пытаюсь читать таблицы улья, используя pyspark, удаленно. В нем говорится об ошибке, что он не может подключиться к клиенту Hive Metastore.

Я прочитал несколько ответов в SO и других источниках, в основном это были конфигурации, но ни один из них не смог решить, почему я не могу подключиться удаленно. Я прочитал документацию и заметил, что без внесения изменений в любой файл конфигурации мы можем подключить спарк с hive. Примечание: я перенаправил порт компьютера, на котором работает hive, и сделал его доступным для localhost:10000. Я даже подключил то же самое с помощью presto и смог выполнить запросы на hive.

Код: 

from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession, HiveContext
SparkContext.setSystemProperty("hive.metastore.uris", "thrift://localhost:9083")
sparkSession = (SparkSession
                .builder
                .appName('example-pyspark-read-and-write-from-hive')
                .enableHiveSupport()
                .getOrCreate())
data = [('First', 1), ('Second', 2), ('Third', 3), ('Fourth', 4), ('Fifth', 5)]
df = sparkSession.createDataFrame(data)
df.write.saveAsTable('example')

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

Абстрактная ошибка: 

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<stdin>", line 2, in <module>
  File "/usr/local/spark/python/pyspark/sql/readwriter.py", line 775, in saveAsTable
    self._jwrite.saveAsTable(name)
  File "/usr/local/spark/python/lib/py4j-0.10.7-src.zip/py4j/java_gateway.py", line 1257, in __call__
  File "/usr/local/spark/python/pyspark/sql/utils.py", line 69, in deco
    raise AnalysisException(s.split(': ', 1)[1], stackTrace)
pyspark.sql.utils.AnalysisException: 'java.lang.RuntimeException: java.lang.RuntimeException: Unable to instantiate org.apache.hadoop.hive.ql.metadata.SessionHiveMetaStoreClient;'

Я запустил команду: 

ssh -i ~/.ssh/id_rsa_sc -L 9000:A.B.C.D:8080 -L 9083:E.F.G.H:9083 -L 10000:E.F.G.H:10000 ubuntu@I.J.K.l

Когда я проверяю порты 10000 и 9083 с помощью команд: 

aviral@versinator:~/testing-spark-hive$ nc -zv localhost 10000
Connection to localhost 10000 port [tcp/webmin] succeeded!
aviral@versinator:~/testing-spark-hive$ nc -zv localhost 9083
Connection to localhost 9083 port [tcp/*] succeeded!

После запуска скрипта я получаю следующую ошибку: 

Caused by: java.net.UnknownHostException: ip-172-16-1-101.ap-south-1.compute.internal
    ... 45 more

1 Ответ

0 голосов
/ 27 марта 2019

Смысл в том, чтобы сохранить конфиги улья при создании самого сеанса spark. 

sparkSession = (SparkSession
                .builder
                .appName('example-pyspark-read-and-write-from-hive')
                .config("hive.metastore.uris", "thrift://localhost:9083", conf=SparkConf())
                .enableHiveSupport()
                .getOrCreate()
                )

Следует отметить, что никаких изменений в настройках spark не требуется, даже такие серверные сервисы, как AWS Glue, могут иметьтакие соединения.

Для полного кода: 

from pyspark import SparkContext, SparkConf
from pyspark.conf import SparkConf
from pyspark.sql import SparkSession, HiveContext
"""
SparkSession ss = SparkSession
.builder()
.appName(" Hive example")
.config("hive.metastore.uris", "thrift://localhost:9083")
.enableHiveSupport()
.getOrCreate();
"""

sparkSession = (SparkSession
                .builder
                .appName('example-pyspark-read-and-write-from-hive')
                .config("hive.metastore.uris", "thrift://localhost:9083", conf=SparkConf())
                .enableHiveSupport()
                .getOrCreate()
                )
data = [('First', 1), ('Second', 2), ('Third', 3), ('Fourth', 4), ('Fifth', 5)]
df = sparkSession.createDataFrame(data)
# Write into Hive
#df.write.saveAsTable('example')

df_load = sparkSession.sql('SELECT * FROM example')
df_load.show()
print(df_load.show())
...