From CSVs to Tables: Infer Schema Data Types From Raw Spreadsheets

import pandas as pd

csv = 'data/fake.csv'

workers_df = pd.read_csv(csv, header=0, encoding='utf-8')
meta = workers_df.info(verbose=True)
print(meta)

<class 'pandas.core.frame.DataFrame'>
RangeIndex: 22 entries, 0 to 21
Data columns (total 11 columns):
id 14 non-null float64
initiated 14 non-null object
hiredate 14 non-null object
email 14 non-null object
firstname 14 non-null object
lastname 14 non-null object
title 14 non-null object
department 14 non-null object
location 14 non-null object
country 14 non-null object
type 14 non-null object
dtypes: float64(1), object(10)
memory usage: 2.0+ KB
None

# File location and type
file_location = "/FileStore/tables/fake.csv"
file_type = "csv"

# CSV options
infer_schema = "true"
first_row_is_header = "true"
delimiter = ","

df = spark.read.format(file_type) \
  .option("inferSchema", infer_schema) \
  .option("header", first_row_is_header) \
  .option("sep", delimiter) \
  .load(file_location)

display(df)

df:pyspark.sql.dataframe.DataFrame
id:integer
initiated:timestamp
hiredate:string
email:string
firstname:string
lastname:string
title:string
department:string
location:string
country:string
type:string

py4j.protocol.Py4JJavaError: An error occurred while calling o43.count.
: java.lang.IllegalArgumentException: Unsupported class file major version 55
        at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:166)
        at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:148)
        at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:136)
        at org.apache.xbean.asm6.ClassReader.<init>(ClassReader.java:237)
        at org.apache.spark.util.ClosureCleaner$.getClassReader(ClosureCleaner.scala:49)
        at org.apache.spark.util.FieldAccessFinder$$anon$3$$anonfun$visitMethodInsn$2.apply(ClosureCleaner.scala:517)
        at org.apache.spark.util.FieldAccessFinder$$anon$3$$anonfun$visitMethodInsn$2.apply(ClosureCleaner.scala:500)
        at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:733)
        at scala.collection.mutable.HashMap$$anon$1$$anonfun$foreach$2.apply(HashMap.scala:134)
        at scala.collection.mutable.HashMap$$anon$1$$anonfun$foreach$2.apply(HashMap.scala:134)

        at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:236)
        at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:40)
        at scala.collection.mutable.HashMap$$anon$1.foreach(HashMap.scala:134)
        at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:732)
        at org.apache.spark.util.FieldAccessFinder$$anon$3.visitMethodInsn(ClosureCleaner.scala:500)
        at org.apache.xbean.asm6.ClassReader.readCode(ClassReader.java:2175)
        at org.apache.xbean.asm6.ClassReader.readMethod(ClassReader.java:1238)
        at org.apache.xbean.asm6.ClassReader.accept(ClassReader.java:631)

import csv
from tableschema import Table

data = 'data/fake.csv'
schema = infer(data, limit=500, headers=1, confidence=0.85)
print(schema)

{
  "fields": [{
    "name": "id",
    "type": "integer",
    "format": "default"
  }, {
    "name": "initiated",
    "type": "string",
    "format": "default"
  }, {
    "name": "hiredate",
    "type": "date",
    "format": "default"
  }, {
    "name": "email",
    "type": "string",
    "format": "default"
  }, {
    "name": "firstname",
    "type": "string",
    "format": "default"
  }, {
    "name": "lastname",
    "type": "string",
    "format": "default"
  }, {
    "name": "title",
    "type": "string",
    "format": "default"
  }, {
    "name": "department",
    "type": "string",
    "format": "default"
  }, {
    "name": "location",
    "type": "string",
    "format": "default"
  }, {
    "name": "country",
    "type": "string",
    "format": "default"
  }, {
    "name": "type",
    "type": "string",
    "format": "default"
  }],
  "missingValues": [""]
}

from sqlalchemy import create_engine
import config
import pandas as pd
import psycopg2
from tableschema import Table, infer, Schema
from functions.recursivejson import extract_values
from sqlalchemy.types import Integer, Text, Date

class CreateTablesFromCSVs:
    """Infer a table schema from a CSV."""

    __uri = config.PG_URI
    __engine = create_engine(__uri, convert_unicode=True, echo=True)
    __data = 'data/fake.csv'

    @classmethod
    def get_data(cls):
        """Pull latest data."""
        test_df = pd.read_csv(cls.__data, header=0, encoding='utf-8')
        return test_df

    @classmethod
    def get_schema_from_csv(cls, csv):
        """Infers schema from CSV."""
        table = Table(csv)
        table.infer(limit=500, confidence=0.55)
        schema = table.schema.descriptor
        names = cls.get_column_names(schema, 'name')
        datatypes = cls.get_column_datatypes(schema, 'type')
        schema_dict = dict(zip(names, datatypes))
        return schema_dict

    @classmethod
    def get_column_names(cls, schema, key):
        """Get names of columns."""
        names = extract_values(schema, key)
        return names

    @classmethod
    def get_column_datatypes(cls, schema, key):
        """Convert schema to recognizable by SQLAlchemy."""
        values = extract_values(schema, key)
        for i, value in enumerate(values):
            if value == 'integer':
                values[i] = Integer
            elif value == 'string':
                values[i] = Text
            elif value == 'date':
                values[i] = Date
        return values

    @classmethod
    def create_new_table(cls, data, schema):
          """Create new table from CSV and generated schema."""
          workday_table.to_sql('faketable',
                               con=cls.__engine,
                               schema='testschema',
                               if_exists='replace',
                               chunksize=300,
                               dtype=schema)

data = CreateTablesFromCSVs.get_schema_from_csv()
schema = CreateTablesFromCSVs.get_schema_from_csv(data)
CreateTablesFromCSVs.create_new_table(data, schema)