Spark: операция concat для mysql

Question

У меня таблица клиентов в mysql, как показано ниже

 mysql> describe customers;
 +-------------------+--------------+------+-----+---------+----------------+
 | Field             | Type         | Null | Key | Default | Extra          |
 +-------------------+--------------+------+-----+---------+----------------+
 | customer_id       | int(11)      | NO   | PRI | NULL    | auto_increment |
 | customer_fname    | varchar(45)  | NO   |     | NULL    |                |
 | customer_lname    | varchar(45)  | NO   |     | NULL    |                |
 | customer_email    | varchar(45)  | NO   |     | NULL    |                |
 | customer_password | varchar(45)  | NO   |     | NULL    |                |
 | customer_street   | varchar(255) | NO   |     | NULL    |                |
 | customer_city     | varchar(45)  | NO   |     | NULL    |                |
 | customer_state    | varchar(45)  | NO   |     | NULL    |                |
 | customer_zipcode  | varchar(45)  | NO   |     | NULL    |                |
 +-------------------+--------------+------+-----+---------+----------------+

Я читаю эту таблицу в spark (версия 1.6), как показано ниже

 scala> val prob1 = sqlContext.read.jdbc("jdbc:mysql://localhost:3306/retail_db","customers",prop)

, а затем с некоторыми изменениями пытаюсьзаписать обратно в другую таблицу в mysql

 scala> prob1.select('customer_id,concat('customer_fname,'customer_lname)).where('customer_id > 12430).write.jdbc("jdbc:mysql://localhost:3306/retail_db","customer_seg3",prop)

com.mysql.jdbc.exceptions.jdbc4.MySQLSyntaxErrorException: You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near '(customer_fname,customer_lname) TEXT NOT NULL)' at line 1
at sun.reflect.GeneratedConstructorAccessor58.newInstance(Unknown Source)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)

ПРИМЕЧАНИЕ. Если я попробую то же самое без оператора concat, все будет нормально в mysql и от spark до mysql

 mysql> select customer_id,concat(customer_fname,customer_lname) from 
        customers where customer_id > 12430;
        +-------------+---------------------------------------+
        | customer_id | concat(customer_fname,customer_lname) |
        +-------------+---------------------------------------+
        |       12431 | MaryRios                              |
        |       12432 | AngelaSmith                           |
        |       12433 | BenjaminGarcia                        |
        |       12434 | MaryMills                             |
        |       12435 | LauraHorton                           |
        +-------------+---------------------------------------+   
        5 rows in set (0.01 sec)

Может кто-нибудь подтвердить, чточто здесь не так?

Также я попробовал другой способ. Сохранял результат prob1 в другом DF, как показано ниже, и писал, что DF возвращается в таблицу mysql.

  scala> val prob2 = prob1.select('customer_id,concat('customer_fname,'customer_lname)).where('customer_id > 12430)
        prob2: org.apache.spark.sql.DataFrame = [customer_id: int, concat(customer_fname,customer_lname): string]

        scala> prob2.printSchema
        root
         |-- customer_id: integer (nullable = false)
         |-- concat(customer_fname,customer_lname): string (nullable = false)


         scala> val prob2 =  prob1.select('customer_id,concat('customer_fname,'customer_lname)).where('customer_id > 12430).toDF("id","name")
         prob2: org.apache.spark.sql.DataFrame = [id: int, name: string]

         scala> prob2.write.jdbc("jdbc:mysql://localhost:3306/retail_db","cust_2",prop)

Так как же, второйметод работал с той же логикой и конструкциями

С уважением, Sumit