PythonSpark:需要从文件列执行配置单元查询

Posted

技术标签:

【中文标题】PythonSpark:需要从文件列执行配置单元查询【英文标题】:PythonSpark: need to execute hive queries from file columns 【发布时间】:2019-08-01 19:08:09 【问题描述】:

我有一个包含如下行的文件(文件名:sample.csv

Id,Query
T1012,"Select * from employee_dim limit 100"
T1212,"Select * from department_dim limit 100"
T1231,"Select dept_number,location,dept_name from locations"

我需要遍历这个文件 (sample.csv) 并获取第二列 ("query"),在 hive 数据库中运行它并获取结果,然后将其保存到一个名为 T1012_result.csv 的新文件中,同样对所有行执行此操作。

你能帮忙吗?

我尝试通过 spark 读取文件并将其转换为列表,然后使用无效的 sparksession 执行 SQL 查询。

from pyspark.sql import SparkSession,HiveContext

spark=SparkSession.builder.enableHiveSupport().getOrCreate()
spark.sql("use sample")
input=spark.read.csv("sample.csv")
#input.select('_c1').show()

import pandas as pd

a=input.toPandas().values.tolist()
for i in a :
   print i[1]
   spark.sql('pd.DataFrame(i)')

【问题讨论】:

【参考方案1】:

更新:火花

file_path="file:///user/vikrant/inputfiles/multiquery.csv"
df=spark.read.format("com.databricks.spark.csv").option("header", "true").load(file_path)

+---+-------------------------------+
|id |query                          |
+---+-------------------------------+
|1  |select * from exampledate      |
|2  |select * from test             |
|3  |select * from newpartitiontable|
+---+-------------------------------+

def customFunction(row):
    for row in df.rdd.collect():
        item=(row[1])
        filename=(row[0])
        query=""
        query+=str(item)
        newdf=spark.sql(query)
        savedataframe(newdf,filename)

def savedataframe(newdf,filename):
    newdf.coalesce(1).write.csv("/user/dev/hadoop/external/files/file_" + filename + ".csv")

customFunction(df)

drwxr-xr-x   - vikct001 hdfs          0 2019-08-02 11:49 /user/dev/hadoop/external/files/file_1.csv
drwxr-xr-x   - vikct001 hdfs          0 2019-08-02 11:49 /user/dev/hadoop/external/files/file_2.csv
drwxr-xr-x   - vikct001 hdfs          0 2019-08-02 11:49 /user/dev/hadoop/external/files/file_3.csv

更新:使用熊猫 我在 sql server 上的测试表很少,我正在将它们读入 pandas 数据框,正如您在问题中提到的那样,并将查询结果保存到每个不同的文件中,并重命名为数据框的第一列:

import pandas as pd
import pyodbc
from pandas import DataFrame


connection = pyodbc.connect('Driver=ODBC Driver 13 for SQL Server;SERVER=yourservername;DATABASE=some_db;UID=username;PWD=password')
cursor = connection.cursor()

data=[['1','select * from User_Stage_Table'],['2','select * from User_temp_Table']]
df=pd.DataFrame(data,columns=['id','query'])


def get_query(df):
    a=df.values.tolist()
    for i in a:
        query=i[1]    #reading second column value as query
        filename=i[0] #reading first column value as filename
        write_query(query,filename) #calling write_query function 

def write_query(query,filename):
    df=pd.read_sql_query(query,connection)
    df.to_csv(outfile_location+filename+".txt",sep=',',encoding='utf-8',index=None,mode='a')

get_query(df)  #calling get_query function to build the query
out_file_location='G:\Testing\OutputFile\outfile'

您的输出文件名为:

outfile1.txt #这将有表User_Stage_Table的数据

outfile2.txt #这将有表User_temp_Table'的数据

如果这能解决您的问题或进一步面临任何问题,请告诉我。

【讨论】:

神奇的工作维克兰特。

以上是关于PythonSpark:需要从文件列执行配置单元查询的主要内容,如果未能解决你的问题,请参考以下文章

选择时如何从配置单元视图中丢弃分区列?

有条件地在其他列上从配置单元表中获取最新列值

如何将 json 字符串数据类型列转换为配置单元中的映射数据类型列?

从具有多个分区列的配置单元表中获取最新数据

WPS或者EXCEL如何找出2列数据重复数据

如何读取指定的单元格值并将文件从 excel 转换为电子表格?