Why is input_file_name() empty for S3 catalog sources in pyspark?

user_246606 0
Viewed 7791

I'm trying to get the input file name (or path) for every file loaded through an S3 data catalog in AWS Glue.

I've read in a few places that input_file_name() should provide this information (though caveated that this only works when calling from_catalog and not from_options, which I believe I am!).

So the code below seems like it should work, but always returns empty values for every input_file_name.

import sys

from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from pyspark.sql.functions import input_file_name


args = getResolvedOptions(sys.argv, ['JOB_NAME', 'TempDir'])
sc = SparkContext()
gc = GlueContext(sc)
spark = gc.spark_session


job = Job(gc)
job.init(args['JOB_NAME'], args)


# Get the source frame from the Glue Catalog, which describes files in S3
fm_source = gc.create_dynamic_frame.from_catalog(
    database='database_name',
    table_name='table_name',
    transformation_ctx='fm_source',
)

df_source = fm_source.toDF().withColumn('input_file_name', input_file_name())
df_source.show(5)

Resulting output:

+-------------+---------------+
|other_columns|input_file_name|
+-------------+---------------+
|           13|               |
|           33|               |
|           53|               |
|           73|               |
|           93|               |
+-------------+---------------+

Is there another way that I should be creating the frame that ensures input_file_name() is populated? I've now tried to build a source frame through create_dynamic_frame.from_catalog, create_dynamic_frame.from_options and getSource().getFrame(), but I get the same result of an empty input_file_name column for each.

aws-glue pyspark apache-spark amazon-web-services amazon-s3
7 Answers

I believe this to be impossible when using the groupFiles option, given that behind the scenes Glue is concatenating files to create the optimal number of inputs. Thus, the concept of input_file_name does not make sense in this context, given the original file path is no longer a direct input.

However, the docs are slightly misleading in the sense that even for inputs fewer than 50,000 files, not explicitly disabling the option will trigger Glue to concatenate inputs depending on their file size. In our case, we have thousands of tiny input files (<1 MB) causing this behaviour.

You can easily verify this, by explicitly disabling the grouping (note this will have a severe performance impact for scenarios similar to ours:

ds_s3 = self.gc.getSource(
    connection_type='s3',
    paths=paths,
    groupFiles='none',
)
fm_s3 = ds_s3.getFrame()

Of course, it is better not to have to depend on the input state or context, and so we ended up writing an AWS Lambda triggered on S3 PUT which writes metadata (including the filename and path) into the file itself.

user_2466060

I also add my experience, in my case I received an empty result because calling the cache() method.

For example:

import pyspark.sql.functions as F

df = spark.read.json("/my/folder/test.json")
df.cache()
df = df.withColumn("input_file_name", F.input_file_name())

df.show()

I receive

+-------------+---------------+
|other_columns|input_file_name|
+-------------+---------------+
|           13|               |
|           33|               |
|           53|               |
|           73|               |
|           93|               |
+-------------+---------------+

But if I remove the line df.cache() the column input_file_name shows correctly the input filenames.

A workaround might be calling F.input_file_name() before caching.

user_47250740

I found another weird behavior. When I add a limit() before input_file_path id does not work either.

This code has empty filename

df_raw = (spark
          .read
          .option("delimiter", ";")
          .csv(filepath,header = "true",inferSchema=True)      
          .select("COL1","COL2")
          .limit(1000)
          )

df_transform = (df_raw
          .withColumn("filename", f.input_file_name())
          )

This code works

df_raw = (spark
          .read
          .option("delimiter", ";")
          .csv(filepath,header = "true",inferSchema=True)      
          .select("COL1","COL2")
          )

df_transform = (df_raw
          .withColumn("filename", f.input_file_name())
          .limit(1000)
          )

Tooked me some time to figure that out, because I tried to speed up debuging with reading only a few rows.

user_110550420

I faced the same issues as Vzzarr. When I created the input_file_path column after calling cache on the dataframe, the file path was empty. But When I created the input_file_path before calling cache, it worked.

user_57932950

If all the above still couldn't work, add this..

spark.catalog.clearCache()

to clear all cache in the cluster.

user_50881450

This happened to me because I had a deprecated runtime of my databricks cluster (7.6) and I tried it on 7.3 LTS which worked. :)

user_134537220

Sometimes, there are a few edge cases wherein this method, 'input_file_name()', does not work as expected with DynamicFrame. In such cases, you can try using 'attachFilename' option with DynamicFrame to retrieve file path for the records.

Here is an example:

== Code:

dyf6_ = glue_context.create_dynamic_frame.from_options(
    connection_type="s3",
    connection_options={
        "paths": ["s3://awsglue-datasets/examples/highly-partitioned-table/year=2021/month=12/"],
        "recurse":True,
        "groupFiles": "inPartition",
        "groupSize": "1048576"
    },
    format="json",
    format_options={
        "attachFilename": "filenameCol"
    }
)
dyf6_.toDF().show()
user_99882820
Related