如何在Thrift服务器上查看pyspark临时表？

Question

从pyspark.sql导入行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*

到目前为止一切都很好，一切都很好。

在另一个终端上，我初始化spark thrift Server:./sbin/start-thriftServer.sh--hiveconf hive.server2.thrift.port=10001-conf spark.executor.cores=1--master spark://172.18.0.2:7077

服务器似乎正常启动，我可以看到pyspark和thrift服务器作业在我的spark cluster master UI上运行。

然后我使用beeline连接到集群

./bin/Beeline Beeline>！连接jdbc:hive2://172.18.0.2:10001

我想强调的两件有趣的事情是：

>

启动pyspark时，会收到以下警告

警告ObjectStore：6666-在Metastore中找不到版本信息。未启用hive.mastore.schema.verification，因此记录模式1.2.0版

谢谢




Answer 1

如果有人需要在Spark Streaming中这样做，我让它像这样工作。

from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession
from py4j.java_gateway import java_import

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

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

#Order matters! 
java_import(sc._gateway.jvm, "")
sc._gateway.jvm.org.apache.spark.sql.hive.thriftserver.HiveThriftServer2.startWithContext(spark._jwrapped)

#Define schema of json
schema = StructType().add("partnerid", "string").add("sessionid", "string").add("functionname", "string").add("functionreturnstatus", "string")

#load data into spark-structured streaming
df = spark \
      .readStream \
      .format("kafka") \
      .option("kafka.bootstrap.servers", "localhost:9092") \
      .option("subscribe", "rest_logs") \
      .load() \
      .select(from_json(col("value").cast("string"), schema).alias("parsed_value"))

#Print output
query = df.writeStream \
            .outputMode("append") \
            .format("memory") \
            .queryName("view_figures") \
            .start()

query.awaitTermination();

启动后，可以用beehive进行JDBC测试。我无法理解的是，我必须在相同的脚本中启动Thrift服务器。这就是如何开始脚本。

    spark-submit --master local[2] \
--conf "spark.driver.extraClassPath=D:\\Libraries\\m2_repository\\org\\apache\\kafka\\kafka-clients\\2.0.0\\kafka-clients-2.0.0.jar" \
--packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.1 \
"C:\\Temp\\spark_kafka.py"

希望这能帮到别人。顺便说一句，我正处于初步研究阶段，所以不要对我评头论足。