Hi @Lakshay Thanks so much for your reply. I have looked into most of those options and dont see any python code. Its mostly pipeline.transform. Here is the code where it crashes. I feel it should not bring to the driver either for with column or for writing to parquet. so not sure whats wrong. Happy to share the file as well
from pyspark import StorageLevel
from pyspark.sql.functions import lower, col
import datetime
df = df_joined.repartition(768)
df.persist(StorageLevel.MEMORY_AND_DISK)
# Get the current date and time
formatted_date_time = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
print(formatted_date_time)
# Process rxnorm results
rxnorm_result = (
process_rxnorm_results(
rxnorm_pipelineModel.transform(
process_posology_relations(
model_posology.transform(df)
)
)
)
.withColumn('drug_resolution', lower(col('resolutions')[0]))
.withColumn('chunk1', lower(col('chunk1')))
.withColumn('chunk2', lower(col('chunk2')))
)
rxnorm_result.persist(StorageLevel.MEMORY_AND_DISK)
outname_rx_results = f'{rx_norm_path}/rx_norm/rxnorm_part_{formatted_date_time}.parquet'
rxnorm_result.write.format("parquet").save(outname_rx_results)
df.unpersist()
rxnorm_result.unpersist()
