Error in Databricks using "copy into" SQL command to populate table from csv file (string is not converted to integer)

I have a need to insert data into a Databricks table using data from a csv file.

I've uploaded the file to Databricks, but when I try to use it to insert data into the table using copy into, I get an error that the command is not interpreting the column in the csv file as an integer, but as a string and it is not casting that string to an int. The file was uploaded using the REST API.

How do I get Databricks to write these data to the table?

This is the code I'm currently using.

COPY INTO concept
  FROM '/FileStore/tables/prod//ohdsi/demo_cdm/concept/concept.csv'
  FILEFORMAT = CSV
  FORMAT_OPTIONS ('mergeSchema' = 'true',
                  'inferSchema' ='true',
                  'delimiter' = ',',
                  'header' = 'true')
  COPY_OPTIONS ('mergeSchema' = 'true');

This also fails

COPY INTO concept
  FROM '/FileStore/tables/prod//ohdsi/demo_cdm/concept/concept.csv'
  FILEFORMAT = CSV
  FORMAT_OPTIONS (
    'delimiter' = ',',
    'header' = 'true')
;

This is the error message

Error in SQL statement: AnalysisException: Failed to merge fields 'concept_id' and 'concept_id'. Failed to merge incompatible data types IntegerType and StringType

--- EDIT ----------------------------------------------

I've posted a solution that uses Python.

However, it would be great if I had a SQL solution so I could call this programmatically (or if there is a way I can call the Python programmatically from my local machine).

Spark doesn't work well in interpreting schema of csv data on its own, so a workaround would be to create a table first from the csv data giving it a proper well defined schema and then using the copy into command

We were able to resolve this using a Python script (tables are truncated before running this script):

%python
tables = [
'care_site',
'cdm_source',
'vocabulary'
]

target_db = 'demo_cdm'
data_path = '/FileStore/tables/prod/ohdsi/demo_cdm/'

for t in tables:
  tgt_t = t.lower()
  df = spark.sql('SELECT * FROM {db}.{table}'.format(db=target_db, table=tgt_t))
  spark.read.options(delimiter=",", header="True", dateFormat="yyyy-MM-dd")\
          .schema(df.schema)\
          .csv(data_path + t + '/' + t + '.csv')\
          .write.format('delta')\
          .insertInto(target_db + '.' + tgt_t.lower(), overwrite=True)
  spark.sql('REFRESH TABLE {db}.{table}'.format(db=target_db, table=tgt_t))