Pyspark version 3.x, repartition not working as expected for large JSON data

We have a hadoop cluster of two nodes with around 40 cores and 80 GB RAM. We have to simply digest a large multiline JSON into Elastic Search (ES) cluster. The size of json was 120 GB and after bz2 compression, it is reduced to 2 GB only. We have setup following code for data indexing is ES

....

def start_job():

    warehouse_location = abspath('spark-warehouse')
    # Create a spark session
    spark = SparkSession \
        .builder \
        .appName("Python Spark SQL Hive integration example") \
        .config("spark.sql.warehouse.dir", warehouse_location) \
        .enableHiveSupport() \
        .getOrCreate()

    # Configurations
    spark.conf.set("spark.sql.caseSensitive", "true")

    df = spark.read.option("multiline", "true").json(data_path)
    df = df.repartition(20)
    #Tranformations
    df = df.drop("_id")

    df.write.format(
        'org.elasticsearch.spark.sql'
    ).option(
        'es.nodes', ES_Nodes
    ).option(
        'es.port', ES_PORT
    ).option(
        'es.resource', ES_RESOURCE,
    ).save()

if __name__ == '__main__':

    # ES Setting
    ES_Nodes = "hadoop-master"
    ES_PORT = 9200
    ES_RESOURCE = "myIndex/type"

    # Data absolute path
    data_path = "/dss_data/mydata.bz2"

    start_job()
    print("Job has been finished")

The problem is that only one executor is running as total tasks are one. I was expecting, there should be 20 tasks as I have repartition the data to 20. The Spark UI image is given below. Where is the problem. I am running following command to run the job on cluster

spark-submit --class org.apache.spark.examples.SparkPi --jars elasticsearch-spark-30_2.12-7.14.1.jar    --master yarn  --deploy-mode cluster --driver-memory 10g     --executor-memory 4g --num-executors 20 --executor-cores 2 myscript.py

We are using Hadoop and Spark version 3.x. Further, we are also getting following trace in the Hadoop logs

 df.write.format(
  File "/usr/local/leads/spark/python/lib/pyspark.zip/pyspark/sql/readwriter.py", line 1107, in save
  File "/usr/local/leads/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py", line 1304, in __call__
  File "/usr/local/leads/spark/python/lib/pyspark.zip/pyspark/sql/utils.py", line 117, in deco
pyspark.sql.utils.AnalysisException: Since Spark 2.3, the queries from raw JSON/CSV files are disallowed when the
referenced columns only include the internal corrupt record column
(named _corrupt_record by default). For example:
spark.read.schema(schema).json(file).filter($"_corrupt_record".isNotNull).count()
and spark.read.schema(schema).json(file).select("_corrupt_record").show().
Instead, you can cache or save the parsed results and then send the same query.
For example, val df = spark.read.schema(schema).json(file).cache() and then
df.filter($"_corrupt_record".isNotNull).count().