Fabric Notebook Несанкционированная ошибка при чтении файла паркета ⇐ Python

[Anonymous]

Я относительно новичок в написании Python и использовании Fabric в нашей компании. При этом я создал довольно простой скрипт, который записывает в файл паркета, а затем я хочу взять этот файл паркета и записать его в дельта-таблицу. Однако именно здесь у меня возникают проблемы.
ниже приведен мой сценарий. Это отлично записывает файл паркета. Когда я пытаюсь перенести его в дельта-таблицу, возникает ошибка.
Моя первоначальная попытка заключалась в том, чтобы сделать все вместе. Но я получаю ошибки, когда дело доходит до записи паркета в дельта-таблицу. Я пробовал несколько разных конфигураций, но ни одна из них не работает. Даже тот, который работает в другом скрипте, здесь не работает.
import subprocess
import json
import requests
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
from pyspark.sql import SparkSession

access_token = "TOKEN"

# Define your API endpoint and headers
url = "https://api.monday.com/v2"
headers = {
"Authorization": access_token,
"Content-Type": "application/json"
}

# Define your GraphQL query
query = """
query {
users {
created_at
email
account {
name
id
}
}
}
"""

# Send the request
response = requests.post(url, json={'query': query}, headers=headers)

# Check for errors
if response.status_code == 200:
data = response.json()
# print(data)
users = data['data']['users']

# Convert the data to a DataFrame
df = pd.json_normalize(users)

# Write the DataFrame to a Parquet file
# df.to_parquet('users_data.parquet', engine='pyarrow')
# print("Data written to users_data.parquet")
parquet_table_name = "MondayUsers.parquet"
parquet_file_path = "abfss://Files/Monday/"

df.to_parquet(parquet_file_path + parquet_table_name, engine='pyarrow', index=False)
print("Data written to Parquet file")

# Initialize Spark session with Delta Lake configurations
spark = SparkSession.builder \
.appName("FabricNotebook") \
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
.getOrCreate()

# Define the path to your Parquet file
parquet_file_path = parquet_file_path + parquet_table_name

# Read the Parquet file into a Spark DataFrame
try:
#load parquet file to data lake table
from pyspark.sql import SparkSession

# Initialize Spark session
spark = SparkSession.builder.appName("FabricNotebook").getOrCreate()

# Define the path to your Parquet file

# Read the Parquet file into a Spark DataFrame
df = spark.read.parquet(parquet_file_path)

# Define the Delta table name
delta_table_name = "monday.Users"

# Write the Spark DataFrame to the Delta table
df.write.mode("overwrite").option("overwriteSchema", "true").format("delta").saveAsTable(delta_table_name)

except Exception as e:
print(f"Error reading Parquet file or writing to Delta table: {e}")

else:
print(f"Query failed with status code {response.status_code}")


При этом возникает следующая ошибка:

Ошибка чтения файла Parquet или записи в таблицу Delta: произошла ошибка при вызове o6842.toString. Трассировка:
java.lang.IllegalArgumentException: объект не является экземпляром объявления класса
в java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
в java.base /jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
в java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
в java.base /java.lang.reflect.Method.invoke(Method.java:566)
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine .java:374)
в py4j.Gateway.invoke(Gateway.java:282)
в py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
в py4j.commands. CallCommand.execute(CallCommand.java:79)
в py4j.GatewayConnection.run(GatewayConnection.java:238)
в java.base/java.lang.Thread.run(Thread.java:829)
Я тогда подумал взять и поместить в другой раздел и просто читать с паркета и писать в дельта-таблицу. Но тогда я становлюсь "не авторизованным"
import pandas as pd

parquet_table_name = "MondayUsers.parquet"
parquet_file_path = "abfss://Files/Monday/"

parquet_file = parquet_file_path + parquet_table_name

parquet_df = pd.read_parquet(parquet_file)

spark_df = spark.createDataFrame(parquet_df)

print(spark_df)

# Define the Delta table name
delta_table_name = "monday.Users"

spark_df.write.format("delta").mode("overwrite").saveAsTable(delta_table_name)


Подробнее здесь: https://stackoverflow.com/questions/791 ... rquet-file