<jupyter_start><jupyter_text>POC PYTHON -  Pyspark<jupyter_code>pip install pyspark
import pyspark as pk
import pandas as pd
#https://runawayhorse001.github.io/LearningApacheSpark/pyspark.pdf
print(pk.__version__)
from configparser import ConfigParser

config = ConfigParser()
config.read('config.ini')
sqlsUrl = config['SQLSERVER']['host']
username = config['SQLSERVER']['username']
password = config['SQLSERVER']['password']
port = config['SQLSERVER']['port']

from configparser import ConfigParser

config = ConfigParser()
config.read('config.ini')
database = config['TEST']['database']
sqlsUrl = config['TEST']['host']
username = config['TEST']['username']
password = config['TEST']['password']
port = config['TEST']['port']


# jdbc_url = jdbc:sqlserver://{config.get('mydb', 'host')}:{config.get('mydb', 'port')};database={config.get('mydb', 'database')}"
# connection_properties = {
#     "user": config.get('mydb', 'username'),
#     "password": config.get('mydb', 'password')
# }
# from pyspark.sql import SparkSession

# jars = [
#     "mssql-jdbc-6.4.0.jre8.jar",
# ]
# spark = (SparkSession
#   .builder
#   .appName("PySpark with SQL server")
#   .config("spark.driver.extraClassPath", ":".join(jars))
#   .getOrCreate())  


# .option("driver", "com.microsoft.azure:spark-mssql-connector_2.12:1.2.0")\

#Problème de version JAVA avec pyspark, peut être problématique dans le cadre de BABILOU qui utilise java pour Talend
#Résolu en téléchargeant JDK8
#https://stackoverflow.com/questions/31841509/pyspark-exception-java-gateway-process-exited-before-sending-the-driver-its-po

from pyspark.sql import SparkSession
spark = SparkSession\
    .builder\
    .appName('Connection-Test')\
    .getOrCreate()


qryStr = """ (
SELECT * FROM STG_PROCARE_AR_SCHEDULE
    ) t """

# spark.read.format('jdbc')\
#     .option('url',sqlsUrl)\
#     .option('driver', 'com.microsoft.sqlserver.jdbc.SQLServerDriver')\
#     .option('dbtable', qryStr )\
#     .option("user", user) \
#     .option("password", <PASSWORD>) \
#     .load().show()

# jdbcDF = spark.read \
#     .format("com.microsoft.sqlserver.jdbc.spark") \
#     .option('driver', 'com.microsoft.sqlserver.jdbc.SQLServerDriver')\
#     .option("authentication", "ActiveDirectoryPassword") \
#     .option("encrypt", "true") \
#     .option("hostNameInCertificate", "*.database.windows.net") \
#     .option("url", sqlsUrl+":"+port) \
#     .option("dbtable", qryStr) \
#     .option("user", username) \
#     .option("password", password) \
#     .load()

jdbc_df = spark.read.format("jdbc") \
       .option("url", sqlsUrl+port) \
       .option("driver", "SQL Server Native Client 11.0")\
       .option("dbtable", qryStr) \
       .option("authentication", "ActiveDirectoryPassword") \
       .option("user", username) \
       .option("password", password) \
       .option("encrypt", "true") \
       .option("databaseName", "STG_BABILOU") \
       .option("hostNameInCertificate", "*.database.windows.net") \
       .load()

jdbc_df = spark.read \
        .format("com.microsoft.sqlserver.jdbc.spark") \
        .option("url", url) \
        .option("dbtable", table_name) \
        .option("authentication", "ActiveDirectoryPassword") \
        .option("user", user_name) \
        .option("password", password) \
        .option("encrypt", "true") \
        .option("hostNameInCertificate", "*.database.windows.net") \
        .load()<jupyter_output><empty_output><jupyter_text>Utilisation du csv<jupyter_code>from pyspark.sql import SparkSession
spark = SparkSession\
    .builder\
    .appName('Read_csv')\
    .getOrCreate()
procare = spark.read.options(header=True)\
.options(delimiter=';')\
.options(inferschema='true')\
.csv("C:\\Users\\charles.lavignasse\\Desktop\\POC-PYTHON\\notebooks\\Connection\procare.csv")

etp= spark.read.options(header=True)\
.options(delimiter=';')\
.options(inferschema='true')\
.csv("C:\\Users\\charles.lavignasse\\Desktop\\POC-PYTHON\\notebooks\\Connection\etp.csv")
procare.show()
etp.collect()[0][0] 

if len(etp.collect())==0: 
    print('empty')

etp['MinDailyHour'][1]
spark
for ligne in procare : 
    CellValue =[]
    for ninter in etp :
        if etp['MinDailyHour'][ninter] <= procare['HoursWorked'][ligne] <= etp['MaxDailyHour'][ninter] Then pass
    CellValue =[etp['Correspondance ETP'][ninter]]
    procare['ETP'][ligne]=CellValue<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.18681569343065693, 'fraction_numerical': 0.006614963503649635, 'mean_word_length': 4.034443168771527, 'pattern_counts': {'":': 4, '<': 12, '<?xml version=': 0, '>': 10, 'https://': 2, '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': '554477', 'license': 'CNRI-Python', 'n_tokens_mistral': 1554, 'n_tokens_neox': 1497, 'n_words': 235, 'path': 'Pyspark_Connect.ipynb', 'repo_name': 'charlesLavignasse/POC-Python-ETL', 'size': '4380'}