使用spark sql 和spark streaming进行数据查询，将结果保存在数据库

首你需要创建一个SparkSession对象和一个StreamingContext对象，然后将数据流读入并转换成DataFrame。

接着，你可以使用Spark SQL中的DataFrame API和SQL语句查询数据，并将结果保存到数据库中。下面是一个示例代码：

from pyspark.sql import SparkSession
from pyspark.streaming import StreamingContext
from pyspark.sql.functions import *

# 创建SparkSession对象
spark = SparkSession.builder.appName("query_and_save").getOrCreate()

# 创建StreamingContext对象
ssc = StreamingContext(spark.sparkContext, batchDuration=10)

# 从Kafka读取数据流，假设数据格式为"key,value"
kafka_stream = KafkaUtils.createDirectStream(ssc, [topic], {"metadata.broker.list": brokers})

# 将数据流转换成DataFrame
df = kafka_stream.map(lambda x: x[1]).\
    map(lambda row: row.split(",")).\
    map(lambda row: (row[0], row[1])).\
    toDF(["key", "value"])

# 注册DataFrame为临时表
df.createOrReplaceTempView("my_table")

# 使用Spark SQL查询数据
result = spark.sql("SELECT key, count(*) as count FROM my_table GROUP BY key")

# 将结果保存到数据库
result.write.format("jdbc").options(
    url="jdbc:mysql://localhost:3306/my_database",
    driver="com.mysql.jdbc.Driver",
    dbtable="my_table",
    user="my_username",
    password="my_password").mode("append").save()

在上面的代码中，我假设你使用了Kafka作为数据源，并且数据格式为"key,value"。你可以根据自己的数据源和数据格式进行相应的修改。我也假设你使用了MySQL数据库，你可以根据自己的需求修改数据库相关信息。