In Pyspark how to generate serial number on multiple CSV files maintaining order

Viewed 47
+--------+------+
|Name    | SrlNo|
+--------+------+
|Sweden  | 1    |
|Albania | 2    |
|India   | 3    |
|Iceland | 4    |
|Finland | 5    |
|Denmark | 6    |
|Algeria | 8    |
|Andorra | 9    |
|Norway  | 10   |
+-------+-------|

I have the above data frame. I want to partition and save data into multiple CSV files. I am able to do it by the below glue code, but glue is randomly picking the row and making CSV files like below.

finalCount=dynamicFrame.count()
records_per_file=14701
    partition_count = math.ceil(finalCount / records_per_file)
    if partition_count < 1:
        partition_count = 1

dynamicFrame = dynamicFrame.repartition(partition_count)
    glueContext.write_dynamic_frame.from_options(
    frame=dynamicFrame,
    connection_type="s3",
    connection_options={
        "path": "S3_Path",
        'groupFiles': 'inPartition', 'groupSize': '10485760'
    },
    format="csv",
    format_options={
        "optimizePerformance": True, 
        "separator": ","
        },
    transformation_ctx="AmazonS3_",
)

CSV files

CSV 1
+--------+------+
|Name    | SrlNo|
+--------+------+
|Sweden  | 1    |
|India   | 3    |
|Finland | 5    |
|Denmark | 6    |
|Andorra | 9    |
+-------+-------|


CSV 2
+--------+------+
|Name    | SrlNo|
+--------+------+
|Albania | 2    |
|India   | 3    |
|Iceland | 4    |
|Algeria | 8    |
|Norway  | 10   |
+-------+-------|

My expected output is this.

CSV 1:
+--------+------+
|Name    | SrlNo|
+--------+------+
|Sweden  | 1    |
|Albania | 2    |
|India   | 3    |
|Iceland | 4    |
+-------+-------|
    CSV 2
    +--------+------+
    |Name    | SrlNo|
    +--------+------+
    |Finland | 5    |
    |Denmark | 6    |
    |Algeria | 8    |
    |Andorra | 9    |
    |Norway  | 10   |
    +-------+-------|

I am a beginner in pyspark., appreciate the guidance. My glue version in 3.0, spark version is 3.1, and Python version is 3

1 Answers

You can try creating one column on which you can repartition data

like

dynamicFrame=dynamicFrame.withColumn("repa",F.ceil(F.col("srlno")/partition_count).cast("int"))
dynamicFrame = dynamicFrame.repartition(partition_count,"repa")
Related