jorgecardona/jupyterlabmultilanguages repository overview

def test_mongo_connection(host, port, database, collection, user=None, password=None):
    """
    # Configure MongoDB credentials
    host = "host.docker.internal"  # Replace with the IP address or hostname of your MongoDB server
    port = 27017  # Default port for MongoDB
    database = "spark"  # Name of the database you want to connect to
    collection = "users"  # Name of the collection
    user = "admin"  # Username (optional, if MongoDB is configured with authentication)
    password = "12345678"  # Password (optional, if MongoDB is configured with authentication)
    
    # If you want to use authentication, you need to provide credentials
    # mongodb://admin:12345678@localhost:27017
    
    # test the connection
    test_mongo_connection(host, port, database, collection)
    """
    
    from pymongo import MongoClient
    
    try:
        # Try to connect to the MongoDB server if have an user and password
        client = MongoClient(host, port, username=user, password=password)

        # Try to connect to the MongoDB server if you do not have an user and password
        
        if not user or not password:
            client = MongoClient(host, port)

        # Select the database
        db = client[database]

        # Perform a simple read operation to verify the connection
        # You can use any query here; for example, count_documents({})
        collection_loaded = db[collection]
        result = collection_loaded.find_one()

        if result:
            print("Connection successful. MongoDB server is accessible.", result)
        else:
            print("Connection successful, but no data was found in the database.")
        
        client.close()
    except Exception as e:
        print(f"Error connecting to the database: {e}")
                
def test_postgres_connection(host, port, database, user, password):
    """
    # Configure PostgreSQL credentials
    host = "host.docker.internal"  # Replace with the IP address or hostname of your PostgreSQL server
    port = 5432  # Default port for PostgreSQL
    database = "spark"  # Name of the database you want to connect to
    user = "admin"  # Username
    password = "12345678"  # Password
    
    # test the connection
    test_postgres_connection(host, port, database, user, password)
    """
    import psycopg2
    
    try:
        # Try to connect to the database
        connection = psycopg2.connect(
            host=host,
            port=port,
            database=database,
            user=user,
            password=password
        )
        connection.close()
        print("Connection successful. PostgreSQL server is accessible.")
    except Exception as e:
        print(f"Error connecting to the database: {e}")
                
def test_mysql_connection(host, port, database, user, password):
    """
    # Configure MySQL credentials
    host = "host.docker.internal"  # Replace with the IP address or hostname of your MySQL server
    port = 3306  # Default port for MySQL
    database = "spark"  # Name of the database you want to connect to
    user = "admin"  # Username
    password = "12345678"  # Password
    
    # test the connection
    test_mysql_connection(host, port, database, user, password)
    """    
    import mysql.connector
    
    try:
        # Try to connect to the MySQL database
        connection = mysql.connector.connect(
            host=host,
            port=port,
            database=database,
            user=user,
            password=password
        )
        connection.close()
        print("Connection successful. MySQL server is accessible.")
        
    except Exception as e:
        print(f"Error connecting to the database: {e}")

def get_database_configuration(database_type = 'mysql', host = None, port = None, database = None, table = None, user = None, password = None, input_collection = None, output_collection = None):
    
    from pyspark.sql import SparkSession
    
    databases = {
        'mongodb': {
            'app_name': 'MongoDB_Connector',
            'format_type':'mongo',
            'host': host if database_type == 'mongodb' and host else 'host.docker.internal',
            'port': port if database_type == 'mongodb' and port else 27017,
            'user': user if database_type == 'mongodb' and user else 'admin',
            'password': password if database_type == 'mongodb' and password else '12345678',
            'database': database if database_type == 'mongodb' and database else 'spark',
            'input_collection':  input_collection if database_type == 'mongodb' and table else 'users_in',
            'output_collection': output_collection if database_type == 'mongodb' and table else 'users_out',
            'spark_jars': '/usr/local/spark/jars/mongo-spark-connector_2.13-10.2.0.jar',
            'spark.mongodb.input.uri': f'mongodb://{host}/{database}.{input_collection}' if database_type == 'mongodb' and table else 'mongodb://host.docker.internal.users_in',
            'spark.mongodb.output.uri': f'mongodb://{host}/{database}.{output_collection}' if database_type == 'mongodb' and table else 'mongodb://host.docker.internal.users_out',
            'driver': 'com.mongodb.spark.sql.DefaultSource',
            'spark_session_read': SparkSession.builder.master('local') \
                                                 .appName("MongoDB_Connector") \
                                                 .config("spark.mongodb.input.uri", f'mongodb://{host}/{database}.{input_collection}' if database_type == 'mongodb' and table else 'mongodb://host.docker.internal.users_in') \
                                                 .config("spark.mongodb.input.uri", f'mongodb://{host}/{database}.{output_collection}' if database_type == 'mongodb' and table else 'mongodb://host.docker.internal.users_out') \
                                                 .config("spark.jars", '/usr/local/spark/jars/mongo-spark-connector_2.13-10.2.0.jar') \
                                                 .getOrCreate()
            },
        'postgres': {
            'app_name': 'PostgreSQL_Connector',
            'format_type':'jdbc',
            'host': host if database_type == 'postgres' and host else 'host.docker.internal',
            'port': port if database_type == 'postgres' and port else 5432,
            'user': user if database_type == 'postgres' and user else 'admin',
            'password': password if database_type == 'postgres' and password else '12345678',
            'database': database if database_type == 'postgres' and database else 'spark',
            'table': table if database_type == 'postgres' and table else 'users',
            'schema': 'public',
            'spark_jars': '/usr/local/spark/jars/postgresql-42.6.0.jar',
            'driver': 'org.postgresql.Driver',
            'url': f"jdbc:postgresql://{host}:{port}/{database}" if database_type == 'postgres' and host and port else 'jdbc:postgresql://host.docker.internal:5432/spark',
            'properties': {
                'user': user if database_type == 'postgres' and user else 'admin',
                'password': password if database_type == 'postgres' and password else '12345678',
                'driver': 'org.postgresql.Driver'
            },
            'spark_session_read': SparkSession.builder.master('local').appName("PostgreSQL_Connector").config("spark.jars", '/usr/local/spark/jars/postgresql-42.6.0.jar').getOrCreate()
            },
        'mysql': {
            'app_name': 'MySQL_Connector',
            'format_type':'jdbc',
            'host': host if database_type == 'mysql' and host else 'host.docker.internal',
            'port': port if database_type == 'mysql' and port else 3306,
            'user': user if database_type == 'mysql' and user else 'admin',
            'password': password if database_type == 'mysql' and password else '12345678',
            'database': database if database_type == 'mysql' and database else 'spark',
            'table': table if database_type == 'mysql' and table else 'users',
            'spark_jars': '/usr/local/spark/jars/mysql-connector-j-8.0.33.jar',
            'driver': 'com.mysql.cj.jdbc.Driver',
            'url': f"jdbc:mysql://{host}:{port}/{database}" if database_type == 'mysql' and host and port else 'jdbc:mysql://host.docker.internal:3306/spark',
            'properties': { 
                            'user': user if database_type == 'mysql' and user else 'admin', 
                            'password': password if database_type == 'mysql' and password else '12345678', 
                            'driver': 'com.mysql.cj.jdbc.Driver'
                
            },
            'spark_session_read': SparkSession.builder.master('local').appName("MySQL_Connector").config("spark.jars", '/usr/local/spark/jars/mysql-connector-j-8.0.33.jar').getOrCreate()
            }
    }
    return databases.get(database_type.lower(), databases.get('mysql'))

def insert_data_to_database(database_configuration):
    
    app_name = database_configuration.get('app_name')
    format_type = database_configuration.get('format_type')
    driver = database_configuration.get('driver')
    url = database_configuration.get('url')
    dbtable = database_configuration.get('table')
    user = database_configuration.get('user')
    password = database_configuration.get('password')    
    
    from pyspark.sql import SparkSession
    from pyspark.sql.functions import monotonically_increasing_id
    
    spark_session_write = SparkSession.builder.appName("WriteToDatabase").getOrCreate()

    # Crear un DataFrame de ejemplo
    data = [(1, "Ana"), (2, "Cecilia"), (3, "Nathalie"), (4, "Diana"), (5, "Gabriela"), (6, "Angela"), (7, "Tatiana"), (8, "Lucia"), (9, "Maria")]
    columns = ["Id", "Name"]
    sampleDF = spark_session_write.createDataFrame(data, columns)

    # Agregar una columna 'id' al DataFrame
    sampleDF_with_id = sampleDF.withColumn("id", monotonically_increasing_id())

    sampleDF.write \
        .format(format_type) \
        .option("driver", driver) \
        .option("url", url) \
        .option("dbtable", dbtable) \
        .option("user", user) \
        .option("password", password) \
        .mode("ignore") \
        .mode("append") \
        .save()

    # Detener la sesión de Spark
    spark_session_write.stop()
    
    return 'Records Inserted Successfully in {app_name}'.format(app_name=app_name)

# GET DATABASE CONFIGURATION
mysql_configuration = get_database_configuration(database_type = 'mysql')

# INSERT DATA INTO DATABASE AND CREATES THE TABLE
insert_data_to_database(mysql_configuration)

# GET DATABASE CONFIGURATION
postgres_configuration = get_database_configuration(database_type = 'postgres')

# INSERT DATA INTO DATABASE AND CREATES THE TABLE
insert_data_to_database(postgres_configuration)

def read_data_from_database(database_type = 'mysql', host = None, port = None, database = None, table = None, user = None, password = None, input_collection = None, output_collection = None):
    
    database_configuration = get_database_configuration(database_type = database_type, host = host, port = port, database = database, table = table, user = user, password = password, input_collection = input_collection, output_collection = output_collection)
    
    spark_instance = database_configuration.get('spark_session_read')
    properties = database_configuration.get('properties')
    spark = database_configuration.get('spark_session')
    url = database_configuration.get('url')
    
    table = database_configuration.get('table')
    
    if database_configuration.get('schema'):
        table = f"{database_configuration.get('schema')}.{database_configuration.get('table')}"

    spark_dataframe = spark_instance.read.jdbc(url=url, table=table, properties=properties)
    result = spark_dataframe.toPandas()
    spark_instance.stop()
    
    return result

read_data_from_database(database_type = 'mysql')

read_data_from_database(database_type = 'postgres')