Java JdbcRDD Spark

java-jdbc-rdd-spark

import java.io.Serializable;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;

import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.rdd.JdbcRDD;

import scala.Function0;
import scala.Function1;

import org.apache.spark.api.java.function.Function;
import scala.reflect.ClassTag;
import scala.reflect.ClassTag$;
import scala.runtime.BoxedUnit;

public class MySQLConnector implements Serializable {
	
	static Connection connection = null;
	
	private static class mysqlFunction implements Function1<ResultSet, Object>, Serializable {

		public String call(ResultSet arg0) throws Exception {
			String output = "";	
			for(int i = 1; i <= arg0.getMetaData().getColumnCount(); i++) {
				String ss = arg0.getString(i);
				output = output + "\t" + ss;
				System.out.println(ss);
			}
			
			return output;
			// TODO Auto-generated method stub
		}

		@Override
		public <A> Function1<ResultSet, A> andThen(Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcDD$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcDF$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcDI$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcDJ$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcFD$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcFF$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcFI$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcFJ$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcID$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcIF$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcII$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcIJ$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcJD$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcJF$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcJI$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcJJ$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcVD$sp(
				Function1<BoxedUnit, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcVF$sp(
				Function1<BoxedUnit, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcVI$sp(
				Function1<BoxedUnit, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcVJ$sp(
				Function1<BoxedUnit, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcZD$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcZF$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcZI$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<Object, A> andThen$mcZJ$sp(
				Function1<Object, A> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public Object apply(ResultSet arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public double apply$mcDD$sp(double arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public double apply$mcDF$sp(float arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public double apply$mcDI$sp(int arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public double apply$mcDJ$sp(long arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public float apply$mcFD$sp(double arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public float apply$mcFF$sp(float arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public float apply$mcFI$sp(int arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public float apply$mcFJ$sp(long arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public int apply$mcID$sp(double arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public int apply$mcIF$sp(float arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public int apply$mcII$sp(int arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public int apply$mcIJ$sp(long arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public long apply$mcJD$sp(double arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public long apply$mcJF$sp(float arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public long apply$mcJI$sp(int arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public long apply$mcJJ$sp(long arg0) {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public void apply$mcVD$sp(double arg0) {
			// TODO Auto-generated method stub
			
		}

		@Override
		public void apply$mcVF$sp(float arg0) {
			// TODO Auto-generated method stub
			
		}

		@Override
		public void apply$mcVI$sp(int arg0) {
			// TODO Auto-generated method stub
			
		}

		@Override
		public void apply$mcVJ$sp(long arg0) {
			// TODO Auto-generated method stub
			
		}

		@Override
		public boolean apply$mcZD$sp(double arg0) {
			// TODO Auto-generated method stub
			return false;
		}

		@Override
		public boolean apply$mcZF$sp(float arg0) {
			// TODO Auto-generated method stub
			return false;
		}

		@Override
		public boolean apply$mcZI$sp(int arg0) {
			// TODO Auto-generated method stub
			return false;
		}

		@Override
		public boolean apply$mcZJ$sp(long arg0) {
			// TODO Auto-generated method stub
			return false;
		}

		@Override
		public <A> Function1<A, Object> compose(Function1<A, ResultSet> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcDD$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcDF$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcDI$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcDJ$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcFD$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcFF$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcFI$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcFJ$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcID$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcIF$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcII$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcIJ$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcJD$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcJF$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcJI$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcJJ$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, BoxedUnit> compose$mcVD$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, BoxedUnit> compose$mcVF$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, BoxedUnit> compose$mcVI$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, BoxedUnit> compose$mcVJ$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcZD$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcZF$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcZI$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public <A> Function1<A, Object> compose$mcZJ$sp(
				Function1<A, Object> arg0) {
			// TODO Auto-generated method stub
			return null;
		}
	}
	
	public static void main(String[] args) throws Exception {
		
		SparkConf sparkConf = new SparkConf().setAppName("JdbcRddTest");
		sparkConf.setMaster("local[2]");
		SparkContext ctx = new SparkContext(sparkConf);

		JdbcRDD<Object> rdd = new JdbcRDD<Object>(ctx,
				new getConnection(), "SELECT * FROM gps_table WHERE ? <= id AND id <= ?",
				1, 3000, 10, new mysqlFunction(), ClassTag$.MODULE$.apply(Object.class));
		
		
		System.out.println("Number of rows :" + rdd.count());


		
	}
	
	
	private static class getConnection implements Function0<Connection>, Serializable{

		@Override
		public Connection apply() {
			try {
				Class.forName("org.gjt.mm.mysql.Driver");
			} catch (ClassNotFoundException e) {
				System.out.println("Where is your MySQL JDBC Driver? "
					+ "Include in your library path!");
				e.printStackTrace();
				return null;
			}
			try {
				connection = DriverManager.getConnection(
						   "jdbc:mysql://localhost:3306/sigmoid"
						   , "akhld", "pass");
			} catch (SQLException e) {
				System.out.println("Connection Failed!");
				e.printStackTrace();
				return null;
			}
			return connection;
		}

		@Override
		public byte apply$mcB$sp() {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public char apply$mcC$sp() {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public double apply$mcD$sp() {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public float apply$mcF$sp() {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public int apply$mcI$sp() {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public long apply$mcJ$sp() {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public short apply$mcS$sp() {
			// TODO Auto-generated method stub
			return 0;
		}

		@Override
		public void apply$mcV$sp() {
			// TODO Auto-generated method stub
			
		}

		@Override
		public boolean apply$mcZ$sp() {
			// TODO Auto-generated method stub
			return false;
		}
		
	}

}