Посмотрите на свой код. Вы определили DB_HOST и запросили DB_SERVER. DB_USER и DB_USERNAME Используйте этот код
define("DB_SERVER", "localhost");
define("DB_USER", "root");
define("DB_PASSWORD", "");
define("DB_DATABASE", "databasename");
$connect = mysqli_connect(DB_SERVER , DB_USER, DB_PASSWORD, DB_DATABASE);
можно использовать flatMap для получения столбца желаемый кадр данных сразу
df=df.withColumn('udf_results',udf)
df4=df.select('udf_results').rdd.flatMap(lambda x:x).toDF(schema=your_new_schema)