sql_connector.py

import json
from abc import ABCMeta, abstractmethod

from pyspark import SQLContext, SparkContext
from pyspark.sql import DataFrame


class AbstractSQLConnector(metaclass=ABCMeta):
    @classmethod
    def load_config(cls, vendor: str, config_instance: str, config_path: str) -> (str, str):
        try:
            with open(config_path, 'r') as f:
                configs = json.load(f)
                config = configs[config_instance]

                props = {
                    "user": config['user'],
                    "password": config['password'],
                    "driver": config['driver']
                }

                jdbc_url = \
                    "jdbc:" + vendor + \
                    "://" + config['host'] + \
                    ":" + config['port'] + \
                    "/" + config['database']

                return jdbc_url, props

        except IOError:
            print("File " + config_path + " not found.")

    @abstractmethod
    def read(self, sqlc: SQLContext, table: str) -> DataFrame:
        pass

    @abstractmethod
    def write(self, df: DataFrame, table: str) -> None:
        pass


class MySQLConnector(AbstractSQLConnector):
    def __init__(self, config_instance: str, config_path: str):
        self.read_url, self.props = AbstractSQLConnector.load_config("mysql", config_instance, config_path)
        self.write_url = self.read_url

    def read(self, sqlc: SQLContext, table: str) -> DataFrame:
        return sqlc.read.jdbc(url=self.read_url, table=table, properties=self.props)

    def write(self, df: DataFrame, table: str, mode: str = "append") -> None:
        df.write.mode(mode).jdbc(url=self.write_url, table=table, properties=self.props)


def main():
    sc = SparkContext("local[*]", "SparkRecipesJob").getOrCreate()
    sqlc = SQLContext(sc)
    sqlc.setConf("spark.jars.packages", "mysql:mysql-connector-java:5.1.40")

    mysql = MySQLConnector("mysqlconf", 'db/datastore.json')

    stations_df = mysql.read(sqlc, "stations")
    stations_df.printSchema()
    stations_df.show()


# spark-submit --packages mysql:mysql-connector-java:5.1.40 db/sql_connector.py
if __name__ == '__main__':
    main()

my python class (rewritten from scala) for reading tables as spark dataframe just with conn.read(spark, table)