<jupyter_start><jupyter_text>Abstractive Summarization of Earnings Call Transcripts<jupyter_code># Install sparknlp
!pip install --upgrade sparknlp
import os
import sparknlp

from pyspark.sql.types import *
from pyspark.ml import Pipeline

os.getcwd()<jupyter_output><empty_output><jupyter_text>> To get started you will need to include the JDBC driver for you particular database on the spark classpath. For example, to connect to postgres from the Spark Shell you would run the following command:<jupyter_code>from pyspark.sql import SparkSession

# Initiating a Spark NLP Session 
spark = SparkSession.builder \
    .appName("Spark NLP")\
    .master("local[4]")\
    .config("spark.python.worker.memory","6G") \
    .config("spark.driver.memory","24G")\
    .config("spark.driver.maxResultSize", "0") \
    .config("spark.kryoserializer.buffer.max", "2000M")\
    .config("spark.serializer","org.apache.spark.serializer.KryoSerializer") \
    .config("spark.jars.packages", "com.johnsnowlabs.nlp:spark-nlp_2.12:3.1.1")\
    .config("spark.driver.extraClassPath", "./mysql-connector-java-8.0.25.jar")\
    .config("spark.worker.cleanup.enabled",True)\
    .getOrCreate()

# // Loading data from a JDBC source
jdbcDF = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:mysql://sem_mysql:3306/test_db") \
    .option("dbtable", "earningscall") \
    .option("user", "es_user") \
    .option("password", "<PASSWORD>") \
    .load()

print(jdbcDF.columns)
jdbcDF.show(5)
from pyspark.sql.functions import concat,lit

folder_path = os.getcwd()
folder_name =  "earnings_call_transcripts"
file_path = os.path.join(folder_path, folder_name, '')

jdbcDF = jdbcDF.select(concat(lit('file:'),lit(file_path),jdbcDF.filepath) \
               .alias("local_filepath"),'companyid', 'ticker', 'company_name', 'filepath', 'transcript')

# jdbcDF.select(concat_ws('/',lit(local_path),lit(folder_name),jdbcDF.filepath)
#               .alias("local_filepath"),'companyid', 'ticker', 'company_name', 'filepath', 'transcript').select('local_filepath').toPandas()['local_filepath'][0]

jdbcDF.printSchema()
from pyspark.sql.types import *
from pyspark.sql.functions import *    
# from pyspark.sql.functions import monotonically_increasing_id 

file_path = "./earnings_call_transcripts/"
texts = spark.sparkContext.wholeTextFiles(file_path,use_unicode=True)

schema = StructType([
    StructField('path', StringType()),
    StructField('raw_text', StringType())
#     StructField('text_len',IntegerType())
])

# Add index and Handle escape sequence of \x00 before tokenizing the document
textdf = spark.createDataFrame(texts, schema=schema).persist() \
              .withColumn("index", monotonically_increasing_id()) \
              .withColumn('text', regexp_replace('raw_text', '\x00|\n', ''))

# Add text length to the table
import pyspark.sql.functions as f

textdf = textdf.withColumn("original_text_length",f.length(col("text")))

textdf.printSchema()

textdf.show(5)
# Join tables from mysql with pyspark table
jdbcJoinedTable = jdbcDF.join(textdf, jdbcDF.local_filepath == textdf.path, 'left').select(jdbcDF.local_filepath,jdbcDF.companyid,jdbcDF.ticker,jdbcDF.company_name,jdbcDF.filepath, textdf.text,textdf.original_text_length) 
jdbcJoinedTable.show()
import gc
gc.collect()
spark.catalog.clearCache()

from sparknlp.annotator import *
from sparknlp.common import *
from sparknlp.base import *
from pyspark.ml import Pipeline

assembler = DocumentAssembler()\
    .setInputCol('text')\
    .setOutputCol('doc')
t5 = T5Transformer() \
    .pretrained("t5_small") \
    .setTask("summarize:")\
    .setMaxOutputLength(200)\
    .setInputCols(["doc"]) \
    .setOutputCol("summaries")

pipeline = Pipeline().setStages([assembler, t5])
results = pipeline.fit(jdbcJoinedTable).transform(jdbcJoinedTable)
results.printSchema()
# Summary of Earnings call Transcripts for Broadcom Q
results.filter(col("ticker") == "AMAT").select(['summaries.result']).toPandas()['result'][0][0]<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1490264603095357, 'fraction_numerical': 0.009735396904643036, 'mean_word_length': 5.0804248861911985, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '627259', 'license': 'MIT', 'n_tokens_mistral': 1410, 'n_tokens_neox': 1341, 'n_words': 243, 'path': 'Abstractive Text Summarization using Pyspark.ipynb', 'repo_name': 'prabhupavitra/Text-Summarization-PySpark', 'size': '4005'}