Hoping to get some help on how to insert data from Spark to Ignite.
The test cases in the build are too trivial to help.
Here is Try 3:
It fails with this error at the moment:
diagnostics: User class threw exception: javax.cache.CacheException: class
org.apache.ignite.internal.processors.query.IgniteSQLException: Failed to
parse query: insert into STATUS (_key, id, statusid, statusdesc) values
(777, "uniqueid", 777, "description")
object SparkLoadIgniteData2 {
def main(args: Array[String]): Unit = {
val sparkConf = new
SparkConf().setMaster("yarn").setAppName("IgniteTest")
val sparkContext = new SparkContext(sparkConf)
val tcpDiscoverySpi: TcpDiscoverySpi = new TcpDiscoverySpi
val ipFinder: TcpDiscoveryVmIpFinder = new TcpDiscoveryVmIpFinder
val addressList: util.List[String] = new util.ArrayList[String]
addressList.add("XX.XXX.XX.XXXX:47500")
ipFinder.setAddresses(addressList)
tcpDiscoverySpi.setIpFinder(ipFinder)
tcpDiscoverySpi.setLocalAddress("XX.XXX.XX.XXX")
val igniteConf : IgniteConfiguration = new
IgniteConfiguration().setClientMode(true).setDiscoverySpi(tcpDiscoverySpi)
val igniteContext = new IgniteContext(sparkContext, () => igniteConf)
val igniteCache = igniteContext.fromCache("STATUS")
igniteCache.sql("insert into STATUS (_key, id, statusid, statusdesc)
values (777, \"uniqueid\", 777, \"description\")")
}
** I also tried to specify the values 'afterwards' in the igniteCache.sql
call and use ?,?,?,? in the values statement, but always got this same parse
error. I also tried single quotes around the values like 'uniqueid' and got
the same error. I also tried a semicolon at the end of the query, and got
the same error.
Try 2:
Using very similar code as above to create the connection to ignite:
Use savePairs or saveValues
case class Status(uniqueId:String, statusId: Int, statusDesc: String)
val newData = List(Status("uniqueId", 777, "a description"))
val newDataRdd = sparkContext.parallelize(newData)
igniteCache.saveValues(newDataRdd, ?????? )
or
igniteCache.savePairs(newDataRdd, ?????? )
I could not figure out how to complete the Scala code and get it to compile
to get this to work.
Try 1:
Use a data frame in Spark to write to Ignite.
........
val aStatus = Seq(Status("3333", 3333, "From Apache Spark!"))
val newDf = sparkSession.createDataFrame(aStatus)
var prop = new Properties();
prop.setProperty("url","jdbc:ignite://XX.XXX.XX.XXX:11211");
prop.setProperty("port", "11211");
newDf.write.mode(SaveMode.Overwrite).jdbc(url, "STATUS", prop);
...or
newDf.write.format("com.databricks.spark.csv").save("dataframefromselect")
This failed to establish a connection to Ignite when trying to write..but no
idea why.
The same program was able to read data from the same Ignite cache earlier on
in the code.
Thanks in advance.
--
View this message in context:
http://apache-ignite-users.70518.x6.nabble.com/Inserting-Data-From-Spark-to-Ignite-tp14937.html
Sent from the Apache Ignite Users mailing list archive at Nabble.com.
