Ethereum JDBC with Spark

As spark provides support of reading from and writing to JDBC data sources, you can use ethereum JDBC driver as one of that kind. Just add ethereum jdbc driver dependency in your application and you are ready to explore.

To know the basics of how to use any JDBC driver with spark, you can refer the below link (referred from spark documentation)
Spark-JDBC to other datasources

Register ethereum JDBC Dialect

You need to register the ethereum JDBC Dialect before any read/write operation. This dialect extends the JdbcDialect provided by spark and maps the ethereum data types to supported spark data types.

you can get the etherem jdbcDialect and register as below-

    JdbcDialects.registerDialect(EthereumDialect)

Reading Data from ethereum tables

Refer below code snippet for reading from ethereum tables and storing as spark dataframe

    JdbcDialects.registerDialect(EthereumDialect);
    val blockDF= spark.read
      .format("jdbc")
      .option("driver", "com.impetus.eth.jdbc.EthDriver")
      .option("url", "jdbc:blkchn:ethereum://ropsten.infura.io/1234")
      .option("dbtable", "block")
      .load()
    blockDF.where("blocknumber > 2256446 and blocknumber < 2256451").show()

Save DataFrame

If you want to transfer ethers to multiple recipients/ multiple transactions to single recipient , you can create DataFrame and initiate all the transactions in a single call.

create DataFrame with transaction details ( one transaction each row) -

Refer below code snippet -

    import spark.implicits._
    val txnDF= Seq(
      Transaction("a76cd046cf6089fe2adcf1680fcede500e44bacd", 0.1, "ether", true),
      Transaction("8144c67b144A408ABC989728e32965EDf37Adaa1", 0.2, "ether", true),
      Transaction("a76cd046cf6089fe2adcf1680fcede500e44bacd", 0.3, "ether", true)).toDS().toDF()

    case class Transaction(toAddress: String, value: Double, unit: String, asyc: Boolean)

Refer below code snippet for initiating all the transactions stored in created DataFrame.

    val prop = new Properties();
    prop.put("KEYSTORE_PATH",
      "<KEYSTORE LOCATION OF SENDER>");
    prop.put("KEYSTORE_PASSWORD", "SENDER'S KEYSTORE PASSWORD");

    JdbcDialects.registerDialect(EthereumDialect);

    txnDF.write.mode(SaveMode.Append).option("driver", 
    "com.impetus.eth.jdbc.EthDriver").jdbc("jdbc:blkchn:ethereum://ropsten.infura.io/1234", "transaction", prop);

Note- Always use SaveMode.Append