Create CustomForEachSink.java

Author: abulbasar

src/main/java/com/example/CustomForEachSink.java

@@ -0,0 +1,69 @@
+packagecom.example;
+
+importjava.sql.Connection;
+importjava.sql.DriverManager;
+importjava.sql.PreparedStatement;
+importjava.sql.SQLException;
+
+importorg.apache.spark.sql.ForeachWriter;
+importorg.apache.spark.sql.Row;
+
+/*
+
+Prepare the mysql database
+1. create database
+create database iot;
+2. Create a table inside iot database
+create table tags(tag varchar(255) primary key not null, count bigint not null);
+
+Add mysql jdbc as maven dependency to the project.
+
+*/
+
+publicclassCustomForEachSinkextendsForeachWriter<Row>{
+
+privatestaticfinallongserialVersionUID = 1L;
+privateConnectionconn = null; 
+privatePreparedStatementstatement = null;
+
+
+@Override
+publicbooleanopen(longpartitionId, longversion){
+try{
+Class.forName("com.mysql.jdbc.Driver");
+conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/iot","root","cloudera");
+Stringsql = "insert into tags (tag, count) value (?, ?) ON DUPLICATE KEY UPDATE count=?";
+statement = conn.prepareStatement(sql);
+ } catch (Exceptione){
+// TODO Auto-generated catch block
+e.printStackTrace();
+ }
+returntrue;
+ }
+
+@Override
+publicvoidprocess(Rowrow){
+Stringtag = row.getString(0);
+Longcount = row.getLong(1);
+try{
+statement.setString(1, tag);
+statement.setLong(2, count);
+statement.setLong(3, count);
+statement.execute();
+ } catch (SQLExceptione){
+// TODO Auto-generated catch block
+e.printStackTrace();
+ }
+System.out.println(tag + ":" + count);
+ }
+
+@Override
+publicvoidclose(ThrowableerrorOrNull){
+try{
+conn.close();
+ } catch (SQLExceptione){
+// TODO Auto-generated catch block
+e.printStackTrace();
+ }
+ }
+ }