Попытка преобразовать объект «org. apache .spark. sql .DataFrame» в pandas dataframe приводит к ошибке «name 'dataframe' is not defined» в Databricks - PullRequest
Новая
       118

Попытка преобразовать объект «org. apache .spark. sql .DataFrame» в pandas dataframe приводит к ошибке «name 'dataframe' is not defined» в Databricks

0 голосов
/ 29 мая 2020

Я пытаюсь запросить базу данных SQL через соединение jdb c в блоках данных и сохранить результаты запроса как pandas фрейм данных. Все методы, которые я могу найти для этого в Интернете, включают его сохранение как тип объекта Spark, сначала с использованием кода Scala, а затем преобразование его в pandas. Я пробовал для ячейки 1: 

%scala
val df_table1 = sqlContext.read.format("jdbc").options(Map(
    ("url" -> "jdbc:sqlserver://myserver.database.windows.net:1433;database=mydb"),
    ("dbtable" -> "(select top 10 * from myschema.table) as table"),
    ("user" -> "user"),
    ("password" -> "password123"),
    ("driver" -> "com.microsoft.sqlserver.jdbc.SQLServerDriver"))
).load()

, что дает: 

df_table1: org.apache.spark.sql.DataFrame = [var1: int, var2: string ... 50 more fields]

Отлично! Но когда я пытаюсь преобразовать его в pandas df в ячейке 2, чтобы я мог его использовать: 

import numpy as np
import pandas as pd 

result_pdf = df_table1.select("*").toPandas()

print(result_pdf)

Он генерирует сообщение об ошибке: 

NameError: name 'df_table1' is not defined

Как мне успешно преобразовать этот объект в фрейм данных pandas или, альтернативно, есть ли способ запросить базу данных SQL через соединение jdb c с использованием кода python без необходимости использовать Scala вообще (мне не особенно нравится Scala синтаксис и предпочел бы избежать его, если это вообще возможно)?

1 Ответ

0 голосов
/ 30 мая 2020

Я предполагаю, что вы намерены запросить SQL, используя python, и в этом случае следующий запрос будет работать. 

%python
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession
conf = SparkConf()
conf.setMaster("local").setAppName("My app")
sc = SparkContext.getOrCreate(conf=conf)
spark = SparkSession(sc)
database = "YourDBName"
table = "[dbo].[YourTabelName]"
user = "SqlUser"
password  = "SqlPassword"

DF1 = spark.read.format("jdbc") \
    .option("url", f"jdbc:sqlserver://YourAzureSql.database.windows.net:1433;databaseName={database};") \
    .option("dbtable", table) \
    .option("user", user) \
    .option("password", password) \
    .option("driver", "com.microsoft.sqlserver.jdbc.SQLServerDriver") \
    .load()
DF1.show()

table = "[dbo].[someOthertable]"

DF2 = spark.read.format("jdbc") \
    .option("url", f"jdbc:sqlserver://YourAzureSql.database.windows.net:1433;databaseName={database};") \
    .option("dbtable", table) \
    .option("user", user) \
    .option("password", password) \
    .option("driver", "com.microsoft.sqlserver.jdbc.SQLServerDriver") \
    .load()
DF2.show()

Finaldf = DF1.join(DF2,(DF1.Prop_0 == DF2.prop_0),how="inner").select(DF1.Prop_0,DF1.Prop_1,DF2.Address)
Finaldf.show()
...