# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# +
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession
from pyspark.sql import *
from pyspark.sql.types import *
from pyspark.sql.functions import udf
from pyspark.sql.functions import *
from pyspark.sql.window import Window
NoneType = type(None)
import os
import socket
import hashlib
import string

import time
from osgeo import ogr
import geopandas as gpd
from pyspark.sql import SparkSession
from sedona.register import SedonaRegistrator
from sedona.utils import SedonaKryoRegistrator, KryoSerializer


# +
def createMd5(text):
    return hashlib.md5(text.encode('utf-8')).hexdigest()
md5Udf= udf(lambda z: createMd5(z),StringType())

def clean_lower(text):
    sentence = text.translate(str.maketrans('', '', '!"#$%&\'()*+,./:;<=>?@[\\]^`{|}~-_”“«»‘')).lower()
    return " ".join(sentence.split())
cleanLowerUdf= udf(lambda z: clean_lower(z),StringType())

def get_site_from_url(text):
    return text.split("/")[2]
getUrl= udf(lambda z: get_site_from_url(z),StringType())    

# -

minio_ip = socket.gethostbyname('minio')
spark = SparkSession. \
    builder. \
    appName("Python Spark S3"). \
    config("spark.serializer", KryoSerializer.getName). \
    config("spark.executor.memory", "80g"). \
    config("spark.driver.memory", "80g"). \
    config('spark.dirver.maxResultSize', '5g'). \
    config("spark.kryo.registrator", SedonaKryoRegistrator.getName). \
    config('spark.hadoop.fs.s3a.endpoint', 'http://'+minio_ip+':9000'). \
    config("spark.hadoop.fs.s3a.access.key", "minio-access-key"). \
    config("spark.hadoop.fs.s3a.secret.key", "minio-secret-key"). \
    config('spark.hadoop.fs.s3a.impl', 'org.apache.hadoop.fs.s3a.S3AFileSystem'). \
    config('spark.jars.packages',
           'org.apache.sedona:sedona-python-adapter-3.0_2.12:1.0.0-incubating,org.datasyslab:geotools-wrapper:geotools-24.0'). \
    getOrCreate()
SedonaRegistrator.registerAll(spark)

# +

st= StructType([
    StructField("abstract", StringType()),
    StructField("authors", StringType()),
    StructField("image", StringType()),
    StructField("metadata", StringType()),
    StructField("publish_date", TimestampType()),
    StructField("text", StringType()),
    StructField("title", StringType()),
    StructField("url", StringType()),
])
# -

df_news_covid_mexico = spark.read.schema(st).option("timestampFormat", "dd-MM-yyyy").json("s3a://news/covid_mexico/*.json")

df_news_covid_mexico.count()

df_news_covid_mexico.printSchema()

df_news_covid_mexico.show(10)

df_news_covid_mexico_date_text = df_news_covid_mexico.select(md5Udf("url").alias("article_id"),"title","url","publish_date",cleanLowerUdf("text").alias("clean_text"),getUrl("url").alias("site")).filter("length(text) >= 2")

df_news_covid_mexico_date_text.show(15)

df_news_covid_mexico_date_text.count()

df_news_covid_mexico_date_text.select("title").show(15,False)

url = "jdbc:postgresql://postgres/shared"
mode="overwrite"
properties = {
    "user": "shared",
    "password": os.environ['SHARED_PASSWORD']
}

# +
#os.environ['SHARED_PASSWORD']='<PASSWORD>'
# -

df_news_covid_mexico_date_text.write.jdbc(url=url, table="tb_news_covid_mexico_date_text", mode=mode, properties=properties)



StarcoderdataJupyterScriptsDedupFiltered