SaveAsTable error when trying to move table to another path using Spark and Hadoop API

sasson gabai 0
Viewed 26

So I am trying to move a table from one path to another, I get an error saying I got Unclosed character class error and can't figure this out.

This is my code:

import os 
import subprocess
        
from pyspark.sql.functions import input_file_name, regexp_replace from pyspark.sql import SparkSession
        
URI           = sc._gateway.jvm.java.net.URI 
Path          = sc._gateway.jvm.org.apache.hadoop.fs.Path 
FileSystem    = sc._gateway.jvm.org.apache.hadoop.fs.FileSystem 
Configuration = sc._gateway.jvm.org.apache.hadoop.conf.Configuration
        
fs = FileSystem.get(URI("some_url/some_ip"), Configuration())
        
desired_table = "some_table" 
source_db = "some_source" 
target_db = "some_target"
        
        file_list = []
        
        fixed_location = some_source+ "/" + some_table+ "/current"
        
        target_path = fs.listStatus(Path('XXX')); source_folder = fs.listStatus(Path('XXX' + fixed_location))
        
        for file_status in source_folder:
            file_list.append(str(file_status.getPath()))
             spark = SparkSession.builder.master('local').appName('test').getOrCreate()
        
        for file_status in file_list:
            df = spark.read.parquet(file_status)
            df.coalesce(50).write.option("path", target_path).mode('append').format('parquet').saveAsTable(desired_table)

and this is the error:

An error occurred while calling o114.saveAsTable.
: java.io.IOException: Illegal file pattern: Unclosed character class near index 43
[Lorg.apache.hadoop.fs.FileStatus;@38099a5e
                                           ^
    at org.apache.hadoop.fs.GlobFilter.init(GlobFilter.java:71)
    at org.apache.hadoop.fs.GlobFilter.<init>(GlobFilter.java:50)
    at org.apache.hadoop.fs.Globber.glob(Globber.java:192)
    at org.apache.hadoop.fs.FileSystem.globStatus(FileSystem.java:1697)
    at org.apache.spark.deploy.SparkHadoopUtil.globPath(SparkHadoopUtil.scala:244)
    at org.apache.spark.deploy.SparkHadoopUtil.globPathIfNecessary(SparkHadoopUtil.scala:254)
    at org.apache.spark.sql.execution.datasources.DataSource$.org$apache$spark$sql$execution$datasources$DataSource$$checkAndGlobPathIfNecessary(DataSource.scala:707)
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$15.apply(DataSource.scala:389)
    at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$15.apply(DataSource.scala:389)

I am on it for hours, suggestions are welcome !

hdfs pyspark apache-spark python
1 Answers

your target_path = fs.listStatus(Path('XXX')) not in type string, check the document what type of object it's returning pls. the log is showing the error at saveAsTable since spark is executing your code lazily and only at saveAsTable it's triggering the action.

plus i don't understand why you are creating sparksession in a loop

for file_status in source_folder:
            file_list.append(str(file_status.getPath()))
             spark = SparkSession.builder.master('local').appName('test').getOrCreate()

only one sparksession is needed here i think

usermYGG0
Related