package nl.topicus.spannertest.simple;

import java.util.Properties;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.jdbc.JdbcDialect;
import org.apache.spark.sql.jdbc.JdbcDialects;

public class SparkTest
{

	public static void main(String[] args)
	{
		String url = "jdbc:cloudspanner://localhost";
		Properties props = new Properties();
		props.setProperty("Project", "adroit-valve-654321");
		props.setProperty("Instance", "spark-instance");
		props.setProperty("Database", "spark");
		props.setProperty("PvtKeyPath", "C:\\Users\\MyUserName\\Documents\\CloudSpannerKeys\\cloudspanner-key.json");

		testSpark(url, props);
	}

	private static void testSpark(String sparkURL, Properties properties)
	{
		try (SparkSession spark = SparkSession.builder().appName("Java Spark Google Cloud Spanner basic example")
				.master("local").getOrCreate())
		{
			// Register a specific dialect for Cloud Spanner as Cloud Spanner uses backticks (`) for identifier quoting
			JdbcDialects.registerDialect(new JdbcDialect()
			{
				private static final long serialVersionUID = 1L;

				@Override
				public boolean canHandle(String url)
				{
					return url.toLowerCase().startsWith("jdbc:cloudspanner:");
				}

				@Override
				public String quoteIdentifier(String column)
				{
					return "`" + column + "`";
				}
			});
			Dataset<Row> datasetAccount = spark.read().jdbc(sparkURL, "ACCOUNT", properties);
			datasetAccount.show();

			Dataset<Row> datasetEmployee = spark.read().jdbc(sparkURL, "EMPLOYEE", "ID", 1l, 1000l, 10, properties);
			datasetEmployee.show(2000);
		}
	}

}