Apache Spark MySQL JavaRDD.foreachPartition - почему я получаю ClassNotFoundException

Question

Я хотел бы сохранить данные из каждого раздела в базу данных MySQL. Для делать это, я создал класс, который реализует VoidFunction<>:

public class DatabaseSaveFunction implements VoidFunction<Iterator<String>> { 

    /** 
    * 
    */ 
    private static final long serialVersionUID = -7039277486852158360L; 

    public void call(Iterator<String> it) { 
     Connection connect = null; 
     PreparedStatement preparedStatement = null; 

     try { 
      Class.forName("com.mysql.jdbc.Driver"); 
      connect = DriverManager.getConnection("jdbc:mysql://" 
        + "xxx.us-west-2.rds.amazonaws.com" + "/" 
        + "xxx", "xxx", "xxx"); 

      preparedStatement = connect 
        .prepareStatement("insert into testdatabase.test values (default, ?)"); 

      while (it.hasNext()) { 
       String outputElement = it.next(); 
       preparedStatement.setString(1, "" + outputElement.length()); 
       preparedStatement.executeUpdate(); 
      } 

     } catch (ClassNotFoundException e) { 
      e.printStackTrace(); 
     } catch (SQLException e) { 
      e.printStackTrace(); 
     } finally { 
      try { 
       connect.close(); 
       preparedStatement.close(); 
      } catch (SQLException e) { 
       e.printStackTrace(); 
      } 
     } 
    } 
} 

И в моем главном классе метода я звоню:

output.foreachPartition(new DatabaseSaveFunction()); 

Я получаю следующее сообщение об ошибке:

15/05/06 15:34:00 WARN scheduler.TaskSetManager: Lost task 0.0 in stage 1.0 (TID 4, ip-172-31-36-44.us-west-2.compute.internal): java.lang.ClassNotFoundException: DatabaseSaveFunction 
    at java.net.URLClassLoader$1.run(URLClassLoader.java:366) 
    at java.net.URLClassLoader$1.run(URLClassLoader.java:355) 
    at java.security.AccessController.doPrivileged(Native Method) 
    at java.net.URLClassLoader.findClass(URLClassLoader.java:354) 
    at java.lang.ClassLoader.loadClass(ClassLoader.java:425) 
    at java.lang.ClassLoader.loadClass(ClassLoader.java:358) 
    at java.lang.Class.forName0(Native Method) 
    at java.lang.Class.forName(Class.java:274) 

работника log:

15/05/06 15:34:00 ERROR executor.Executor: Exception in task 1.0 in stage 1.0 (TID 5) 
java.lang.ClassNotFoundException: DatabaseSaveFunction 
    at java.net.URLClassLoader$1.run(URLClassLoader.java:366) 
    at java.net.URLClassLoader$1.run(URLClassLoader.java:355) 
    at java.security.AccessController.doPrivileged(Native Method) 
    at java.net.URLClassLoader.findClass(URLClassLoader.java:354) 
    at java.lang.ClassLoader.loadClass(ClassLoader.java:425) 
    at java.lang.ClassLoader.loadClass(ClassLoader.java:358) 
    at java.lang.Class.forName0(Native Method) 
    at java.lang.Class.forName(Class.java:274) 

Может ли кто-нибудь сказать мне, что я делаю неправильно? Я был бы очень благодарен за это.

Answer 1

Экспортируйте внешний класс в jar и добавьте его как sc.addJar ("/ path/x.jar"), где sc - это JavaSparkContext в вашем основном. Тогда вы не получите эту ошибку. Ошибка заключается в том, что программа искры не может найти класс. Кроме того, в искровом режиме 1.3 и выше вы можете просто использовать параметры карты jdbc, а затем использовать load («jdbc», options), чтобы создать кадр данных и загрузить данные из любой СУБД. это действительно удобно. Я не уверен, работает ли этот метод для подключения любых РСУБД к искровым. Скажите, пожалуйста, если у вас есть другие вопросы.