You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@ignite.apache.org by sri hari kali charan Tummala <ka...@gmail.com> on 2019/10/17 17:12:38 UTC
Apache Spark + Ignite Connection Issue
Hi Community,
I am trying to read and write into the Ignite cluster using apache-spark I
am able to do that using JDBC thin client but not native method as
mentioned in several spark + ignite examples.
Right now all the spark + ignite examples launch a local ignite cluster but
I want my code connecting to already existing cluster (client).
Question:-
*How to pass Ignite connection ip and port (10800) 10800 in
example-default.xml ?*
Error:-
*TcpDiscoverySpi: Failed to connect to any address from IP finder (will
retry to join topology every 2000 ms; change 'reconnectDelay' to configure
the frequency of retries): [/3.88.248.113:10800
<http://3.88.248.113:10800>]*
Working (Spark + Ignite using JDBC):-
val df = spark.read
.format("jdbc")
.option("url", "jdbc:ignite:thin://3.88.248.113")
.option("fetchsize",100)
//.option("driver", "org.apache.ignite.IgniteJdbcDriver")
.option("dbtable", "Person").load()
df.printSchema()
df.createOrReplaceTempView("test")
spark.sql("select * from test where id=1").show(10)
spark.sql("select 4,'blah',124232").show(10)
import java.sql.DriverManager
val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
import java.util.Properties
val connectionProperties = new Properties()
connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
spark.sql("select 4 as ID,'blah' as STREET,124232 as
ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
"Person",connectionProperties)
spark.read
.format("jdbc")
.option("url", "jdbc:ignite:thin://3.88.248.113")
.option("fetchsize",100)
.option("dbtable", "Person").load().show(10,false)
Not Working requires a CONFIG file which is example-default.xml:-
val igniteDF = spark.read
.format(FORMAT_IGNITE) //Data source type.
.option(OPTION_TABLE, "person") //Table to read.
.option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
.load()
.filter(col("id") >= 2) //Filter clause.
.filter(col("name") like "%J%") //Another filter clause.
Full Code:- (sparkDSLExample) function fails to connect ignite cluster
which I already have
package com.ignite.examples.spark
import com.ignite.examples.model.Address
import org.apache.ignite.{Ignite, Ignition}
import org.apache.ignite.cache.query.SqlFieldsQuery
import org.apache.ignite.client.{ClientCache, IgniteClient}
import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
import java.lang.{Long => JLong, String => JString}
import org.apache.ignite.cache.query.SqlFieldsQuery
import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE,
OPTION_CONFIG_FILE, OPTION_TABLE}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{SaveMode, SparkSession}
import org.apache.spark.sql.functions.col
object SparkClientConnectionTest {
private val CACHE_NAME = "SparkCache"
private val CONFIG =
"/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
def setupExampleData = {
val cfg2 = new ClientConfiguration().setAddresses("3.88.248.113:10800")
val igniteClient:IgniteClient = Ignition.startClient(cfg2)
System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
val cache:ClientCache[Integer, Address] =
igniteClient.getOrCreateCache(CACHE_NAME)
cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
.setSchema("PUBLIC")).getAll
cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT
EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) )
WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
.setSchema("PUBLIC")).getAll
cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip)
VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco",
"04074").setSchema("PUBLIC")).getAll
cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip)
VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road",
"520003").setSchema("PUBLIC")).getAll
cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip)
VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road",
"1234").setSchema("PUBLIC")).getAll
System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
val data=cache.query(new SqlFieldsQuery("select * from
Person").setSchema("PUBLIC")).getAll
println(data.toString)
}
def sparkDSLExample(implicit spark: SparkSession): Unit = {
println("Querying using Spark DSL.")
println
val igniteDF = spark.read
.format(FORMAT_IGNITE) //Data source type.
.option(OPTION_TABLE, "person") //Table to read.
.option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
.load()
.filter(col("id") >= 2) //Filter clause.
.filter(col("name") like "%J%") //Another filter clause.
println("Data frame schema:")
igniteDF.printSchema() //Printing query schema to console.
println("Data frame content:")
igniteDF.show() //Printing query results to console.
}
def main(args: Array[String]): Unit = {
setupExampleData
//Creating spark session.
implicit val spark = SparkSession.builder()
.appName("Spark Ignite data sources example")
.master("local")
.config("spark.executor.instances", "2")
.getOrCreate()
// Adjust the logger to exclude the logs of no interest.
Logger.getRootLogger.setLevel(Level.ERROR)
Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
//sparkDSLExample
val df = spark.read
.format("jdbc")
.option("url", "jdbc:ignite:thin://3.88.248.113")
.option("fetchsize",100)
//.option("driver", "org.apache.ignite.IgniteJdbcDriver")
.option("dbtable", "Person").load()
df.printSchema()
df.createOrReplaceTempView("test")
spark.sql("select * from test where id=1").show(10)
spark.sql("select 4,'blah',124232").show(10)
import java.sql.DriverManager
val connection =
DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
import java.util.Properties
val connectionProperties = new Properties()
connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
spark.sql("select 4 as ID,'blah' as STREET,124232 as
ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
"Person",connectionProperties)
spark.read
.format("jdbc")
.option("url", "jdbc:ignite:thin://3.88.248.113")
.option("fetchsize",100)
.option("dbtable", "Person").load().show(10,false)
}
}
example-default.xml:-
<?xml version="1.0" encoding="UTF-8"?>
<!--
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
this work for additional information regarding copyright ownership.
The ASF licenses this file to You under the Apache License, Version 2.0
(the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
-->
<!--
Ignite configuration with all defaults and enabled p2p deployment
and enabled events.
-->
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/util
http://www.springframework.org/schema/util/spring-util.xsd">
<bean abstract="true" id="ignite.cfg"
class="org.apache.ignite.configuration.IgniteConfiguration">
<!-- Set to true to enable distributed class loading for
examples, default is false. -->
<property name="peerClassLoadingEnabled" value="true"/>
<!-- Enable task execution events for examples. -->
<property name="includeEventTypes">
<list>
<!--Task execution events-->
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
<!--Cache events-->
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
</list>
</property>
<!-- Explicitly configure TCP discovery SPI to provide list of
initial nodes. -->
<property name="discoverySpi">
<bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
<property name="ipFinder">
<!--
Ignite provides several options for automatic
discovery that can be used
instead os static IP based discovery. For
information on all options refer
to our documentation:
http://apacheignite.readme.io/docs/cluster-config
-->
<!-- Uncomment static IP finder to enable
static-based discovery of initial nodes. -->
<!--<bean
class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
<bean
class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
<property name="addresses">
<list>
<!-- In distributed environment,
replace with actual host IP address. -->
<value>3.88.248.113:10800</value>
</list>
</property>
</bean>
</property>
</bean>
</property>
</bean>
</beans>
--
Thanks & Regards
Sri Tummala
Re: Apache Spark + Ignite Connection Issue
Posted by sri hari kali charan Tummala <ka...@gmail.com>.
so spark + jdbc thin client is the only way to go forward?
Reading:-
val df = spark.read
.format("jdbc")
.option("url", "jdbc:ignite:thin://18.206.247.40")
.option("fetchsize",100)
//.option("driver", "org.apache.ignite.IgniteJdbcDriver")
.option("dbtable", "Person2").load()
df.printSchema()
df.createOrReplaceTempView("test")
spark.sql("select * from test where id=1").show(10)
spark.sql("select 4,'blah',124232").show(10)
Writing:-
import java.sql.DriverManager
val connection = DriverManager.getConnection("jdbc:ignite:thin://18.206.247.40")
import java.util.Properties
val connectionProperties = new Properties()
connectionProperties.put("url", "jdbc:ignite:thin://18.206.247.40")
spark.sql("select 4 as ID,'blah' as STREET,124232 as
ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://18.206.247.40",
"Person2",connectionProperties)
On Fri, Oct 18, 2019 at 12:00 PM sri hari kali charan Tummala <
kali.tummala@gmail.com> wrote:
> unfortunately, tables that are created by spark don't exist in ignite
> when I try to query using sqlline jdbc thin client, so the spark is still
> running locally the tables which are created by spark exists only for that
> session.
>
> did anyone come across this issue? how to resolve it?
>
>
>
>
>
> On Fri, Oct 18, 2019 at 11:32 AM sri hari kali charan Tummala <
> kali.tummala@gmail.com> wrote:
>
>> Hi Stephen/All,
>>
>> got it working somewhat using below, but have an issue table and data
>> which is created using thin client is failing to read using spark but table
>> created by spark can be read using a thin client, that means table created
>> in Ignite using spark are the only ones read using spark in Ignite?
>>
>> example-default.xml
>>
>> <property name="addresses">
>> <list>
>> <!-- In distributed environment, replace with actual host IP address. -->
>> <value>18.206.247.40:10800</value>
>> </list>
>> </property>
>>
>>
>> NotWorking Scala Function:-
>>
>> def readThinClientTableUsingSpark(implicit spark: SparkSession) = {
>> val personDataFrame = spark.read
>> .format(FORMAT_IGNITE)
>> .option(OPTION_CONFIG_FILE, CONFIG)
>> .option(OPTION_TABLE, "person")
>> .load()
>>
>> println()
>> println("Data frame thin connection content:person")
>> println()
>>
>> //Printing content of data frame to console.
>> personDataFrame.show()
>> }
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>>
>> *Exception in thread "main" class org.apache.ignite.IgniteException:
>> Unknown table person at
>> org.apache.ignite.spark.impl.IgniteSQLRelation$$anonfun$schema$2.apply(IgniteSQLRelation.scala:46)
>> at
>> org.apache.ignite.spark.impl.IgniteSQLRelation$$anonfun$schema$2.apply(IgniteSQLRelation.scala:46)
>> at scala.Option.getOrElse(Option.scala:121) at
>> org.apache.ignite.spark.impl.IgniteSQLRelation.schema(IgniteSQLRelation.scala:46)
>> at
>> org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:431)
>> at
>> org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:239)
>> at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:227) at
>> org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:164) at
>> com.ignite.examples.spark.SparkIgniteCleanCode$.readThinClientTableUsingSpark(SparkIgniteCleanCode.scala:154)
>> at
>> com.ignite.examples.spark.SparkIgniteCleanCode$.main(SparkIgniteCleanCode.scala:216)
>> at
>> com.ignite.examples.spark.SparkIgniteCleanCode.main(SparkIgniteCleanCode.scala)*
>>
>> *Full Code:- (step 7 Fails)*
>>
>> package com.ignite.examples.spark
>>
>> import com.ignite.examples.model.Address
>> import org.apache.ignite.{Ignite, Ignition}
>> import org.apache.ignite.cache.query.SqlFieldsQuery
>> import org.apache.ignite.client.{ClientCache, IgniteClient}
>> import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
>> import java.lang.{Long => JLong, String => JString}
>>
>> import com.ignite.examples.spark.SparkClientConnectionTest.{CACHE_NAME, CONFIG}
>> import org.apache.ignite.internal.util.IgniteUtils.resolveIgnitePath
>> import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE, OPTION_CONFIG_FILE, OPTION_CREATE_TABLE_PARAMETERS, OPTION_CREATE_TABLE_PRIMARY_KEY_FIELDS, OPTION_TABLE}
>> import org.apache.log4j.{Level, Logger}
>> import org.apache.spark.sql.{SaveMode, SparkSession}
>>
>> object SparkIgniteCleanCode {
>>
>> private val CACHE_NAME = "SparkCache"
>>
>> private val CONFIG = "/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
>>
>> def setupExampleData = {
>>
>> val cfg2 = new ClientConfiguration().setAddresses("18.206.247.40:10800")
>> val igniteClient:IgniteClient = Ignition.startClient(cfg2)
>>
>> System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
>>
>> val cache:ClientCache[Integer, Address] = igniteClient.getOrCreateCache(CACHE_NAME)
>>
>> cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
>> .setSchema("PUBLIC")).getAll
>>
>> cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) ) WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
>> .setSchema("PUBLIC")).getAll
>>
>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco", "04074").setSchema("PUBLIC")).getAll
>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road", "520003").setSchema("PUBLIC")).getAll
>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road", "1234").setSchema("PUBLIC")).getAll
>>
>> System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
>>
>> val data=cache.query(new SqlFieldsQuery("select * from Person").setSchema("PUBLIC")).getAll
>>
>> println(data.toString)
>> }
>>
>> def sparkReadIgniteWithThinClient(implicit spark: SparkSession)={
>>
>> val df = spark.read
>> .format("jdbc")
>> .option("url", "jdbc:ignite:thin://18.206.247.40")
>> .option("fetchsize",100)
>> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
>> .option("dbtable", "Person").load()
>>
>> df.printSchema()
>>
>> df.createOrReplaceTempView("test")
>>
>> spark.sql("select * from test where id=1").show(10)
>>
>> spark.sql("select 4,'blah',124232").show(10)
>>
>> }
>>
>> def sparkWriteIgniteWithThinClient(implicit spark: SparkSession)={
>>
>> import java.sql.DriverManager
>> val connection = DriverManager.getConnection("jdbc:ignite:thin://18.206.247.40")
>>
>> import java.util.Properties
>> val connectionProperties = new Properties()
>>
>> connectionProperties.put("url", "jdbc:ignite:thin://18.206.247.40")
>>
>> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://18.206.247.40",
>> "Person",connectionProperties)
>>
>> spark.read
>> .format("jdbc")
>> .option("url", "jdbc:ignite:thin://18.206.247.40")
>> .option("fetchsize",100)
>> .option("dbtable", "Person").load().show(10,false)
>>
>> }
>>
>> def writeJSonToIgniteUsingSpark(implicit spark: SparkSession): Unit = {
>>
>> val ignite = Ignition.start(CONFIG)
>>
>> //Load content of json file to data frame.
>> val personsDataFrame = spark.read.json(
>> resolveIgnitePath("/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/resources/person.json").getAbsolutePath)
>>
>> println()
>> println("Json file content:")
>> println()
>>
>> //Printing content of json file to console.
>> personsDataFrame.show()
>>
>> println()
>> println("Writing Data Frame to Ignite:")
>> println()
>>
>> //Writing content of data frame to Ignite.
>> personsDataFrame.write
>> .format(FORMAT_IGNITE)
>> .mode(org.apache.spark.sql.SaveMode.Append)
>> .option(OPTION_CONFIG_FILE, CONFIG)
>> .option(OPTION_TABLE, "json_person")
>> .option(OPTION_CREATE_TABLE_PRIMARY_KEY_FIELDS, "id")
>> .option(OPTION_CREATE_TABLE_PARAMETERS, "template=replicated")
>> .save()
>>
>> println("Done!")
>>
>> println()
>> println("Reading data from Ignite table:")
>> println()
>>
>> val cache = ignite.cache[Any, Any](CACHE_NAME)
>>
>> //Reading saved data from Ignite.
>> val data = cache.query(new SqlFieldsQuery("SELECT id, name, department FROM json_person")).getAll
>>
>> println(data.toString)
>>
>> //data.foreach { row ⇒ println(row.mkString("[", ", ", "]")) }
>> }
>>
>> def readIgniteUsingSpark(implicit spark: SparkSession) = {
>> val json_person = spark.read
>> .format(FORMAT_IGNITE)
>> .option(OPTION_CONFIG_FILE, CONFIG)
>> .option(OPTION_TABLE, "json_person")
>> .load()
>>
>> println()
>> println("Data frame content:json_person")
>> println()
>>
>> //Printing content of data frame to console.
>> json_person.show()
>> }
>>
>>
>> def readThinClientTableUsingSpark(implicit spark: SparkSession) = {
>> val personDataFrame = spark.read
>> .format(FORMAT_IGNITE)
>> .option(OPTION_CONFIG_FILE, CONFIG)
>> .option(OPTION_TABLE, "person")
>> .load()
>>
>> println()
>> println("Data frame thin connection content:person")
>> println()
>>
>> //Printing content of data frame to console.
>> personDataFrame.show()
>> }
>>
>>
>> def main(args: Array[String]): Unit = {
>>
>> println()
>> println("Step 1 setupExampleData:")
>> println()
>>
>> setupExampleData
>>
>> println()
>> println("Step 2 createSparkSession:")
>> println()
>>
>> //Creating spark session.
>> implicit val spark = SparkSession.builder()
>> .appName("Spark Ignite data sources example")
>> .master("local")
>> .config("spark.executor.instances", "2")
>> .getOrCreate()
>>
>> // Adjust the logger to exclude the logs of no interest.
>> Logger.getRootLogger.setLevel(Level.ERROR)
>> Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
>>
>> println()
>> println("Step 3 ReadIgniteWithThinClient of Step1 Data:")
>> println()
>>
>> sparkReadIgniteWithThinClient(spark)
>>
>> println()
>> println("Step 4 sparkWriteIgniteWithThinClient of Step1 Data:")
>> println()
>>
>> sparkWriteIgniteWithThinClient(spark)
>>
>> println()
>> println("Step 5 writeJSonToIgniteUsingSpark Using Spark:")
>> println()
>>
>> writeJSonToIgniteUsingSpark(spark)
>>
>> println()
>> println("Step 6 readIgniteUsingSpark Using Spark:")
>> println()
>>
>> readIgniteUsingSpark(spark)
>>
>> println()
>> println("Step 7 readThinClientTableUsingSpark Using Spark:")
>> println()
>>
>> readThinClientTableUsingSpark(spark)
>>
>>
>> spark.stop()
>>
>>
>> }
>>
>> }
>>
>>
>> example-default.xml
>>
>> <?xml version="1.0" encoding="UTF-8"?>
>>
>> <!--
>> Licensed to the Apache Software Foundation (ASF) under one or more
>> contributor license agreements. See the NOTICE file distributed with
>> this work for additional information regarding copyright ownership.
>> The ASF licenses this file to You under the Apache License, Version 2.0
>> (the "License"); you may not use this file except in compliance with
>> the License. You may obtain a copy of the License at
>>
>> http://www.apache.org/licenses/LICENSE-2.0
>>
>> Unless required by applicable law or agreed to in writing, software
>> distributed under the License is distributed on an "AS IS" BASIS,
>> WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
>> See the License for the specific language governing permissions and
>> limitations under the License.
>> -->
>>
>> <!--
>> Ignite configuration with all defaults and enabled p2p deployment and enabled events.
>> -->
>> <beans xmlns="http://www.springframework.org/schema/beans"
>> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
>> xmlns:util="http://www.springframework.org/schema/util"
>> xsi:schemaLocation="
>> http://www.springframework.org/schema/beans
>> http://www.springframework.org/schema/beans/spring-beans.xsd
>> http://www.springframework.org/schema/util
>> http://www.springframework.org/schema/util/spring-util.xsd">
>> <bean abstract="true" id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
>> <!-- Set to true to enable distributed class loading for examples, default is false. -->
>> <property name="peerClassLoadingEnabled" value="true"/>
>>
>> <!-- Enable task execution events for examples. -->
>> <property name="includeEventTypes">
>> <list>
>> <!--Task execution events-->
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
>>
>> <!--Cache events-->
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
>> </list>
>> </property>
>>
>> <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
>> <property name="discoverySpi">
>> <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
>> <property name="ipFinder">
>> <!--
>> Ignite provides several options for automatic discovery that can be used
>> instead os static IP based discovery. For information on all options refer
>> to our documentation: http://apacheignite.readme.io/docs/cluster-config
>> -->
>> <!-- Uncomment static IP finder to enable static-based discovery of initial nodes. -->
>> <!--<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
>> <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
>> <property name="addresses">
>> <list>
>> <!-- In distributed environment, replace with actual host IP address. -->
>> <value>18.206.247.40:10800</value>
>> </list>
>> </property>
>> </bean>
>> </property>
>> </bean>
>> </property>
>> </bean>
>> </beans>
>>
>>
>> Thanks
>> Sri
>>
>> On Fri, Oct 18, 2019 at 9:23 AM sri hari kali charan Tummala <
>> kali.tummala@gmail.com> wrote:
>>
>>> do you mean communication ports 47100-47200 as mentioned (
>>> https://www.gridgain.com/docs/8.7.6//installation-guide/manual-install-on-ec2)
>>> ? bare in mind I am running my spark job outside ignite ec2 box (MAC PC).
>>>
>>> which option is right in my default.xml file?
>>>
>>> Option 1:-
>>>
>>> <property name="addresses">
>>> <list>
>>> <!-- In distributed environment, replace with actual host IP address. -->
>>> <value>3.88.248.113:4 <http://3.88.248.113:10800/>7100..47200</value>
>>> </list>
>>> </property>
>>>
>>>
>>> Option 2:-
>>>
>>> <property name="addresses">
>>> <list>
>>> <!-- In distributed environment, replace with actual host IP address. -->
>>> <value>3.88.248.113:4 <http://3.88.248.113:10800/>7500..47600</value>
>>> </list>
>>> </property>
>>>
>>>
>>> Option 3:-
>>>
>>> <property name="addresses">
>>> <list>
>>> <!-- In distributed environment, replace with actual host IP address. -->
>>> <value>3.88.248.113 <http://3.88.248.113:10800/></value>
>>> </list>
>>> </property>
>>>
>>>
>>>
>>> Thanks
>>>
>>> On Fri, Oct 18, 2019 at 9:18 AM sri hari kali charan Tummala <
>>> kali.tummala@gmail.com> wrote:
>>>
>>>> Hi Stephen ,
>>>>
>>>> do you mean 3.88.248.113: <http://3.88.248.113:10800/>47500..47700
>>>> something like this? or just public ip 3.88.248.113
>>>> <http://3.88.248.113:10800/> I tried all the possibilities none of
>>>> them are getting connected.
>>>>
>>>> Thanks
>>>> Sri
>>>>
>>>> On Fri, Oct 18, 2019 at 6:02 AM Stephen Darlington <
>>>> stephen.darlington@gridgain.com> wrote:
>>>>
>>>>> You’re trying to connect a thick client (the Spark integration) to the
>>>>> thin client port (10800). Your example-default.xml file needs to have the
>>>>> same configuration as your server node(s).
>>>>>
>>>>> Regards,
>>>>> Stephen
>>>>>
>>>>> On 17 Oct 2019, at 18:12, sri hari kali charan Tummala <
>>>>> kali.tummala@gmail.com> wrote:
>>>>>
>>>>> Hi Community,
>>>>>
>>>>> I am trying to read and write into the Ignite cluster using
>>>>> apache-spark I am able to do that using JDBC thin client but not native
>>>>> method as mentioned in several spark + ignite examples.
>>>>>
>>>>> Right now all the spark + ignite examples launch a local ignite
>>>>> cluster but I want my code connecting to already existing cluster (client).
>>>>>
>>>>> Question:-
>>>>> *How to pass Ignite connection ip and port (10800) 10800 in
>>>>> example-default.xml ?*
>>>>>
>>>>> Error:-
>>>>> *TcpDiscoverySpi: Failed to connect to any address from IP finder
>>>>> (will retry to join topology every 2000 ms; change 'reconnectDelay' to
>>>>> configure the frequency of retries): [/3.88.248.113:10800
>>>>> <http://3.88.248.113:10800/>]*
>>>>>
>>>>> Working (Spark + Ignite using JDBC):-
>>>>>
>>>>> val df = spark.read
>>>>> .format("jdbc")
>>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>>> .option("fetchsize",100)
>>>>> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
>>>>> .option("dbtable", "Person").load()
>>>>>
>>>>> df.printSchema()
>>>>>
>>>>> df.createOrReplaceTempView("test")
>>>>>
>>>>> spark.sql("select * from test where id=1").show(10)
>>>>>
>>>>> spark.sql("select 4,'blah',124232").show(10)
>>>>>
>>>>> import java.sql.DriverManager
>>>>> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>>>>>
>>>>> import java.util.Properties
>>>>> val connectionProperties = new Properties()
>>>>>
>>>>> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>>>>>
>>>>> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
>>>>> "Person",connectionProperties)
>>>>>
>>>>> spark.read
>>>>> .format("jdbc")
>>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>>> .option("fetchsize",100)
>>>>> .option("dbtable", "Person").load().show(10,false)
>>>>>
>>>>>
>>>>> Not Working requires a CONFIG file which is example-default.xml:-
>>>>>
>>>>> val igniteDF = spark.read
>>>>> .format(FORMAT_IGNITE) //Data source type.
>>>>> .option(OPTION_TABLE, "person") //Table to read.
>>>>> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
>>>>> .load()
>>>>> .filter(col("id") >= 2) //Filter clause.
>>>>> .filter(col("name") like "%J%") //Another filter clause.
>>>>>
>>>>>
>>>>> Full Code:- (sparkDSLExample) function fails to connect ignite cluster
>>>>> which I already have
>>>>>
>>>>> package com.ignite.examples.spark
>>>>>
>>>>> import com.ignite.examples.model.Address
>>>>> import org.apache.ignite.{Ignite, Ignition}
>>>>> import org.apache.ignite.cache.query.SqlFieldsQuery
>>>>> import org.apache.ignite.client.{ClientCache, IgniteClient}
>>>>> import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
>>>>> import java.lang.{Long => JLong, String => JString}
>>>>>
>>>>> import org.apache.ignite.cache.query.SqlFieldsQuery
>>>>> import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE, OPTION_CONFIG_FILE, OPTION_TABLE}
>>>>> import org.apache.log4j.{Level, Logger}
>>>>> import org.apache.spark.sql.{SaveMode, SparkSession}
>>>>> import org.apache.spark.sql.functions.col
>>>>>
>>>>> object SparkClientConnectionTest {
>>>>>
>>>>> private val CACHE_NAME = "SparkCache"
>>>>>
>>>>> private val CONFIG = "/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
>>>>>
>>>>> def setupExampleData = {
>>>>>
>>>>> val cfg2 = new ClientConfiguration().setAddresses("3.88.248.113:10800")
>>>>> val igniteClient:IgniteClient = Ignition.startClient(cfg2)
>>>>>
>>>>> System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
>>>>>
>>>>> val cache:ClientCache[Integer, Address] = igniteClient.getOrCreateCache(CACHE_NAME)
>>>>>
>>>>> cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
>>>>> .setSchema("PUBLIC")).getAll
>>>>>
>>>>> cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) ) WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
>>>>> .setSchema("PUBLIC")).getAll
>>>>>
>>>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco", "04074").setSchema("PUBLIC")).getAll
>>>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road", "520003").setSchema("PUBLIC")).getAll
>>>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road", "1234").setSchema("PUBLIC")).getAll
>>>>>
>>>>> System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
>>>>>
>>>>> val data=cache.query(new SqlFieldsQuery("select * from Person").setSchema("PUBLIC")).getAll
>>>>>
>>>>> println(data.toString)
>>>>> }
>>>>>
>>>>> def sparkDSLExample(implicit spark: SparkSession): Unit = {
>>>>> println("Querying using Spark DSL.")
>>>>> println
>>>>>
>>>>>
>>>>> val igniteDF = spark.read
>>>>> .format(FORMAT_IGNITE) //Data source type.
>>>>> .option(OPTION_TABLE, "person") //Table to read.
>>>>> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
>>>>> .load()
>>>>> .filter(col("id") >= 2) //Filter clause.
>>>>> .filter(col("name") like "%J%") //Another filter clause.
>>>>>
>>>>> println("Data frame schema:")
>>>>>
>>>>> igniteDF.printSchema() //Printing query schema to console.
>>>>>
>>>>> println("Data frame content:")
>>>>>
>>>>> igniteDF.show() //Printing query results to console.
>>>>> }
>>>>>
>>>>>
>>>>> def main(args: Array[String]): Unit = {
>>>>>
>>>>> setupExampleData
>>>>>
>>>>> //Creating spark session.
>>>>> implicit val spark = SparkSession.builder()
>>>>> .appName("Spark Ignite data sources example")
>>>>> .master("local")
>>>>> .config("spark.executor.instances", "2")
>>>>> .getOrCreate()
>>>>>
>>>>> // Adjust the logger to exclude the logs of no interest.
>>>>> Logger.getRootLogger.setLevel(Level.ERROR)
>>>>> Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
>>>>>
>>>>> //sparkDSLExample
>>>>>
>>>>>
>>>>> val df = spark.read
>>>>> .format("jdbc")
>>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>>> .option("fetchsize",100)
>>>>> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
>>>>> .option("dbtable", "Person").load()
>>>>>
>>>>> df.printSchema()
>>>>>
>>>>> df.createOrReplaceTempView("test")
>>>>>
>>>>> spark.sql("select * from test where id=1").show(10)
>>>>>
>>>>> spark.sql("select 4,'blah',124232").show(10)
>>>>>
>>>>> import java.sql.DriverManager
>>>>> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>>>>>
>>>>> import java.util.Properties
>>>>> val connectionProperties = new Properties()
>>>>>
>>>>> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>>>>>
>>>>> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
>>>>> "Person",connectionProperties)
>>>>>
>>>>> spark.read
>>>>> .format("jdbc")
>>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>>> .option("fetchsize",100)
>>>>> .option("dbtable", "Person").load().show(10,false)
>>>>>
>>>>> }
>>>>>
>>>>> }
>>>>>
>>>>>
>>>>> example-default.xml:-
>>>>>
>>>>> <?xml version="1.0" encoding="UTF-8"?>
>>>>>
>>>>> <!--
>>>>> Licensed to the Apache Software Foundation (ASF) under one or more
>>>>> contributor license agreements. See the NOTICE file distributed with
>>>>> this work for additional information regarding copyright ownership.
>>>>> The ASF licenses this file to You under the Apache License, Version 2.0
>>>>> (the "License"); you may not use this file except in compliance with
>>>>> the License. You may obtain a copy of the License at
>>>>>
>>>>> http://www.apache.org/licenses/LICENSE-2.0
>>>>>
>>>>> Unless required by applicable law or agreed to in writing, software
>>>>> distributed under the License is distributed on an "AS IS" BASIS,
>>>>> WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
>>>>> See the License for the specific language governing permissions and
>>>>> limitations under the License.
>>>>> -->
>>>>>
>>>>> <!--
>>>>> Ignite configuration with all defaults and enabled p2p deployment and enabled events.
>>>>> -->
>>>>> <beans xmlns="http://www.springframework.org/schema/beans"
>>>>> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
>>>>> xmlns:util="http://www.springframework.org/schema/util"
>>>>> xsi:schemaLocation="
>>>>> http://www.springframework.org/schema/beans
>>>>> http://www.springframework.org/schema/beans/spring-beans.xsd
>>>>> http://www.springframework.org/schema/util
>>>>> http://www.springframework.org/schema/util/spring-util.xsd">
>>>>> <bean abstract="true" id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
>>>>> <!-- Set to true to enable distributed class loading for examples, default is false. -->
>>>>> <property name="peerClassLoadingEnabled" value="true"/>
>>>>>
>>>>> <!-- Enable task execution events for examples. -->
>>>>> <property name="includeEventTypes">
>>>>> <list>
>>>>> <!--Task execution events-->
>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
>>>>>
>>>>> <!--Cache events-->
>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
>>>>> </list>
>>>>> </property>
>>>>>
>>>>> <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
>>>>> <property name="discoverySpi">
>>>>> <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
>>>>> <property name="ipFinder">
>>>>> <!--
>>>>> Ignite provides several options for automatic discovery that can be used
>>>>> instead os static IP based discovery. For information on all options refer
>>>>> to our documentation: http://apacheignite.readme.io/docs/cluster-config
>>>>> -->
>>>>> <!-- Uncomment static IP finder to enable static-based discovery of initial nodes. -->
>>>>> <!--<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
>>>>> <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
>>>>> <property name="addresses">
>>>>> <list>
>>>>> <!-- In distributed environment, replace with actual host IP address. -->
>>>>> <value>3.88.248.113:10800</value>
>>>>> </list>
>>>>> </property>
>>>>> </bean>
>>>>> </property>
>>>>> </bean>
>>>>> </property>
>>>>> </bean>
>>>>> </beans>
>>>>>
>>>>>
>>>>>
>>>>>
>>>>> --
>>>>> Thanks & Regards
>>>>> Sri Tummala
>>>>>
>>>>>
>>>>>
>>>>>
>>>>
>>>> --
>>>> Thanks & Regards
>>>> Sri Tummala
>>>>
>>>>
>>>
>>> --
>>> Thanks & Regards
>>> Sri Tummala
>>>
>>>
>>
>> --
>> Thanks & Regards
>> Sri Tummala
>>
>>
>
> --
> Thanks & Regards
> Sri Tummala
>
>
--
Thanks & Regards
Sri Tummala
Re: Apache Spark + Ignite Connection Issue
Posted by sri hari kali charan Tummala <ka...@gmail.com>.
unfortunately, tables that are created by spark don't exist in ignite
when I try to query using sqlline jdbc thin client, so the spark is still
running locally the tables which are created by spark exists only for that
session.
did anyone come across this issue? how to resolve it?
On Fri, Oct 18, 2019 at 11:32 AM sri hari kali charan Tummala <
kali.tummala@gmail.com> wrote:
> Hi Stephen/All,
>
> got it working somewhat using below, but have an issue table and data
> which is created using thin client is failing to read using spark but table
> created by spark can be read using a thin client, that means table created
> in Ignite using spark are the only ones read using spark in Ignite?
>
> example-default.xml
>
> <property name="addresses">
> <list>
> <!-- In distributed environment, replace with actual host IP address. -->
> <value>18.206.247.40:10800</value>
> </list>
> </property>
>
>
> NotWorking Scala Function:-
>
> def readThinClientTableUsingSpark(implicit spark: SparkSession) = {
> val personDataFrame = spark.read
> .format(FORMAT_IGNITE)
> .option(OPTION_CONFIG_FILE, CONFIG)
> .option(OPTION_TABLE, "person")
> .load()
>
> println()
> println("Data frame thin connection content:person")
> println()
>
> //Printing content of data frame to console.
> personDataFrame.show()
> }
>
>
>
>
>
>
>
>
>
>
>
>
>
> *Exception in thread "main" class org.apache.ignite.IgniteException:
> Unknown table person at
> org.apache.ignite.spark.impl.IgniteSQLRelation$$anonfun$schema$2.apply(IgniteSQLRelation.scala:46)
> at
> org.apache.ignite.spark.impl.IgniteSQLRelation$$anonfun$schema$2.apply(IgniteSQLRelation.scala:46)
> at scala.Option.getOrElse(Option.scala:121) at
> org.apache.ignite.spark.impl.IgniteSQLRelation.schema(IgniteSQLRelation.scala:46)
> at
> org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:431)
> at
> org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:239)
> at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:227) at
> org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:164) at
> com.ignite.examples.spark.SparkIgniteCleanCode$.readThinClientTableUsingSpark(SparkIgniteCleanCode.scala:154)
> at
> com.ignite.examples.spark.SparkIgniteCleanCode$.main(SparkIgniteCleanCode.scala:216)
> at
> com.ignite.examples.spark.SparkIgniteCleanCode.main(SparkIgniteCleanCode.scala)*
>
> *Full Code:- (step 7 Fails)*
>
> package com.ignite.examples.spark
>
> import com.ignite.examples.model.Address
> import org.apache.ignite.{Ignite, Ignition}
> import org.apache.ignite.cache.query.SqlFieldsQuery
> import org.apache.ignite.client.{ClientCache, IgniteClient}
> import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
> import java.lang.{Long => JLong, String => JString}
>
> import com.ignite.examples.spark.SparkClientConnectionTest.{CACHE_NAME, CONFIG}
> import org.apache.ignite.internal.util.IgniteUtils.resolveIgnitePath
> import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE, OPTION_CONFIG_FILE, OPTION_CREATE_TABLE_PARAMETERS, OPTION_CREATE_TABLE_PRIMARY_KEY_FIELDS, OPTION_TABLE}
> import org.apache.log4j.{Level, Logger}
> import org.apache.spark.sql.{SaveMode, SparkSession}
>
> object SparkIgniteCleanCode {
>
> private val CACHE_NAME = "SparkCache"
>
> private val CONFIG = "/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
>
> def setupExampleData = {
>
> val cfg2 = new ClientConfiguration().setAddresses("18.206.247.40:10800")
> val igniteClient:IgniteClient = Ignition.startClient(cfg2)
>
> System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
>
> val cache:ClientCache[Integer, Address] = igniteClient.getOrCreateCache(CACHE_NAME)
>
> cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
> .setSchema("PUBLIC")).getAll
>
> cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) ) WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
> .setSchema("PUBLIC")).getAll
>
> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco", "04074").setSchema("PUBLIC")).getAll
> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road", "520003").setSchema("PUBLIC")).getAll
> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road", "1234").setSchema("PUBLIC")).getAll
>
> System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
>
> val data=cache.query(new SqlFieldsQuery("select * from Person").setSchema("PUBLIC")).getAll
>
> println(data.toString)
> }
>
> def sparkReadIgniteWithThinClient(implicit spark: SparkSession)={
>
> val df = spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://18.206.247.40")
> .option("fetchsize",100)
> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
> .option("dbtable", "Person").load()
>
> df.printSchema()
>
> df.createOrReplaceTempView("test")
>
> spark.sql("select * from test where id=1").show(10)
>
> spark.sql("select 4,'blah',124232").show(10)
>
> }
>
> def sparkWriteIgniteWithThinClient(implicit spark: SparkSession)={
>
> import java.sql.DriverManager
> val connection = DriverManager.getConnection("jdbc:ignite:thin://18.206.247.40")
>
> import java.util.Properties
> val connectionProperties = new Properties()
>
> connectionProperties.put("url", "jdbc:ignite:thin://18.206.247.40")
>
> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://18.206.247.40",
> "Person",connectionProperties)
>
> spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://18.206.247.40")
> .option("fetchsize",100)
> .option("dbtable", "Person").load().show(10,false)
>
> }
>
> def writeJSonToIgniteUsingSpark(implicit spark: SparkSession): Unit = {
>
> val ignite = Ignition.start(CONFIG)
>
> //Load content of json file to data frame.
> val personsDataFrame = spark.read.json(
> resolveIgnitePath("/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/resources/person.json").getAbsolutePath)
>
> println()
> println("Json file content:")
> println()
>
> //Printing content of json file to console.
> personsDataFrame.show()
>
> println()
> println("Writing Data Frame to Ignite:")
> println()
>
> //Writing content of data frame to Ignite.
> personsDataFrame.write
> .format(FORMAT_IGNITE)
> .mode(org.apache.spark.sql.SaveMode.Append)
> .option(OPTION_CONFIG_FILE, CONFIG)
> .option(OPTION_TABLE, "json_person")
> .option(OPTION_CREATE_TABLE_PRIMARY_KEY_FIELDS, "id")
> .option(OPTION_CREATE_TABLE_PARAMETERS, "template=replicated")
> .save()
>
> println("Done!")
>
> println()
> println("Reading data from Ignite table:")
> println()
>
> val cache = ignite.cache[Any, Any](CACHE_NAME)
>
> //Reading saved data from Ignite.
> val data = cache.query(new SqlFieldsQuery("SELECT id, name, department FROM json_person")).getAll
>
> println(data.toString)
>
> //data.foreach { row ⇒ println(row.mkString("[", ", ", "]")) }
> }
>
> def readIgniteUsingSpark(implicit spark: SparkSession) = {
> val json_person = spark.read
> .format(FORMAT_IGNITE)
> .option(OPTION_CONFIG_FILE, CONFIG)
> .option(OPTION_TABLE, "json_person")
> .load()
>
> println()
> println("Data frame content:json_person")
> println()
>
> //Printing content of data frame to console.
> json_person.show()
> }
>
>
> def readThinClientTableUsingSpark(implicit spark: SparkSession) = {
> val personDataFrame = spark.read
> .format(FORMAT_IGNITE)
> .option(OPTION_CONFIG_FILE, CONFIG)
> .option(OPTION_TABLE, "person")
> .load()
>
> println()
> println("Data frame thin connection content:person")
> println()
>
> //Printing content of data frame to console.
> personDataFrame.show()
> }
>
>
> def main(args: Array[String]): Unit = {
>
> println()
> println("Step 1 setupExampleData:")
> println()
>
> setupExampleData
>
> println()
> println("Step 2 createSparkSession:")
> println()
>
> //Creating spark session.
> implicit val spark = SparkSession.builder()
> .appName("Spark Ignite data sources example")
> .master("local")
> .config("spark.executor.instances", "2")
> .getOrCreate()
>
> // Adjust the logger to exclude the logs of no interest.
> Logger.getRootLogger.setLevel(Level.ERROR)
> Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
>
> println()
> println("Step 3 ReadIgniteWithThinClient of Step1 Data:")
> println()
>
> sparkReadIgniteWithThinClient(spark)
>
> println()
> println("Step 4 sparkWriteIgniteWithThinClient of Step1 Data:")
> println()
>
> sparkWriteIgniteWithThinClient(spark)
>
> println()
> println("Step 5 writeJSonToIgniteUsingSpark Using Spark:")
> println()
>
> writeJSonToIgniteUsingSpark(spark)
>
> println()
> println("Step 6 readIgniteUsingSpark Using Spark:")
> println()
>
> readIgniteUsingSpark(spark)
>
> println()
> println("Step 7 readThinClientTableUsingSpark Using Spark:")
> println()
>
> readThinClientTableUsingSpark(spark)
>
>
> spark.stop()
>
>
> }
>
> }
>
>
> example-default.xml
>
> <?xml version="1.0" encoding="UTF-8"?>
>
> <!--
> Licensed to the Apache Software Foundation (ASF) under one or more
> contributor license agreements. See the NOTICE file distributed with
> this work for additional information regarding copyright ownership.
> The ASF licenses this file to You under the Apache License, Version 2.0
> (the "License"); you may not use this file except in compliance with
> the License. You may obtain a copy of the License at
>
> http://www.apache.org/licenses/LICENSE-2.0
>
> Unless required by applicable law or agreed to in writing, software
> distributed under the License is distributed on an "AS IS" BASIS,
> WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
> See the License for the specific language governing permissions and
> limitations under the License.
> -->
>
> <!--
> Ignite configuration with all defaults and enabled p2p deployment and enabled events.
> -->
> <beans xmlns="http://www.springframework.org/schema/beans"
> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
> xmlns:util="http://www.springframework.org/schema/util"
> xsi:schemaLocation="
> http://www.springframework.org/schema/beans
> http://www.springframework.org/schema/beans/spring-beans.xsd
> http://www.springframework.org/schema/util
> http://www.springframework.org/schema/util/spring-util.xsd">
> <bean abstract="true" id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
> <!-- Set to true to enable distributed class loading for examples, default is false. -->
> <property name="peerClassLoadingEnabled" value="true"/>
>
> <!-- Enable task execution events for examples. -->
> <property name="includeEventTypes">
> <list>
> <!--Task execution events-->
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
>
> <!--Cache events-->
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
> </list>
> </property>
>
> <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
> <property name="discoverySpi">
> <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
> <property name="ipFinder">
> <!--
> Ignite provides several options for automatic discovery that can be used
> instead os static IP based discovery. For information on all options refer
> to our documentation: http://apacheignite.readme.io/docs/cluster-config
> -->
> <!-- Uncomment static IP finder to enable static-based discovery of initial nodes. -->
> <!--<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
> <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
> <property name="addresses">
> <list>
> <!-- In distributed environment, replace with actual host IP address. -->
> <value>18.206.247.40:10800</value>
> </list>
> </property>
> </bean>
> </property>
> </bean>
> </property>
> </bean>
> </beans>
>
>
> Thanks
> Sri
>
> On Fri, Oct 18, 2019 at 9:23 AM sri hari kali charan Tummala <
> kali.tummala@gmail.com> wrote:
>
>> do you mean communication ports 47100-47200 as mentioned (
>> https://www.gridgain.com/docs/8.7.6//installation-guide/manual-install-on-ec2)
>> ? bare in mind I am running my spark job outside ignite ec2 box (MAC PC).
>>
>> which option is right in my default.xml file?
>>
>> Option 1:-
>>
>> <property name="addresses">
>> <list>
>> <!-- In distributed environment, replace with actual host IP address. -->
>> <value>3.88.248.113:4 <http://3.88.248.113:10800/>7100..47200</value>
>> </list>
>> </property>
>>
>>
>> Option 2:-
>>
>> <property name="addresses">
>> <list>
>> <!-- In distributed environment, replace with actual host IP address. -->
>> <value>3.88.248.113:4 <http://3.88.248.113:10800/>7500..47600</value>
>> </list>
>> </property>
>>
>>
>> Option 3:-
>>
>> <property name="addresses">
>> <list>
>> <!-- In distributed environment, replace with actual host IP address. -->
>> <value>3.88.248.113 <http://3.88.248.113:10800/></value>
>> </list>
>> </property>
>>
>>
>>
>> Thanks
>>
>> On Fri, Oct 18, 2019 at 9:18 AM sri hari kali charan Tummala <
>> kali.tummala@gmail.com> wrote:
>>
>>> Hi Stephen ,
>>>
>>> do you mean 3.88.248.113: <http://3.88.248.113:10800/>47500..47700
>>> something like this? or just public ip 3.88.248.113
>>> <http://3.88.248.113:10800/> I tried all the possibilities none of them
>>> are getting connected.
>>>
>>> Thanks
>>> Sri
>>>
>>> On Fri, Oct 18, 2019 at 6:02 AM Stephen Darlington <
>>> stephen.darlington@gridgain.com> wrote:
>>>
>>>> You’re trying to connect a thick client (the Spark integration) to the
>>>> thin client port (10800). Your example-default.xml file needs to have the
>>>> same configuration as your server node(s).
>>>>
>>>> Regards,
>>>> Stephen
>>>>
>>>> On 17 Oct 2019, at 18:12, sri hari kali charan Tummala <
>>>> kali.tummala@gmail.com> wrote:
>>>>
>>>> Hi Community,
>>>>
>>>> I am trying to read and write into the Ignite cluster using
>>>> apache-spark I am able to do that using JDBC thin client but not native
>>>> method as mentioned in several spark + ignite examples.
>>>>
>>>> Right now all the spark + ignite examples launch a local ignite cluster
>>>> but I want my code connecting to already existing cluster (client).
>>>>
>>>> Question:-
>>>> *How to pass Ignite connection ip and port (10800) 10800 in
>>>> example-default.xml ?*
>>>>
>>>> Error:-
>>>> *TcpDiscoverySpi: Failed to connect to any address from IP finder (will
>>>> retry to join topology every 2000 ms; change 'reconnectDelay' to configure
>>>> the frequency of retries): [/3.88.248.113:10800
>>>> <http://3.88.248.113:10800/>]*
>>>>
>>>> Working (Spark + Ignite using JDBC):-
>>>>
>>>> val df = spark.read
>>>> .format("jdbc")
>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>> .option("fetchsize",100)
>>>> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
>>>> .option("dbtable", "Person").load()
>>>>
>>>> df.printSchema()
>>>>
>>>> df.createOrReplaceTempView("test")
>>>>
>>>> spark.sql("select * from test where id=1").show(10)
>>>>
>>>> spark.sql("select 4,'blah',124232").show(10)
>>>>
>>>> import java.sql.DriverManager
>>>> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>>>>
>>>> import java.util.Properties
>>>> val connectionProperties = new Properties()
>>>>
>>>> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>>>>
>>>> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
>>>> "Person",connectionProperties)
>>>>
>>>> spark.read
>>>> .format("jdbc")
>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>> .option("fetchsize",100)
>>>> .option("dbtable", "Person").load().show(10,false)
>>>>
>>>>
>>>> Not Working requires a CONFIG file which is example-default.xml:-
>>>>
>>>> val igniteDF = spark.read
>>>> .format(FORMAT_IGNITE) //Data source type.
>>>> .option(OPTION_TABLE, "person") //Table to read.
>>>> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
>>>> .load()
>>>> .filter(col("id") >= 2) //Filter clause.
>>>> .filter(col("name") like "%J%") //Another filter clause.
>>>>
>>>>
>>>> Full Code:- (sparkDSLExample) function fails to connect ignite cluster
>>>> which I already have
>>>>
>>>> package com.ignite.examples.spark
>>>>
>>>> import com.ignite.examples.model.Address
>>>> import org.apache.ignite.{Ignite, Ignition}
>>>> import org.apache.ignite.cache.query.SqlFieldsQuery
>>>> import org.apache.ignite.client.{ClientCache, IgniteClient}
>>>> import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
>>>> import java.lang.{Long => JLong, String => JString}
>>>>
>>>> import org.apache.ignite.cache.query.SqlFieldsQuery
>>>> import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE, OPTION_CONFIG_FILE, OPTION_TABLE}
>>>> import org.apache.log4j.{Level, Logger}
>>>> import org.apache.spark.sql.{SaveMode, SparkSession}
>>>> import org.apache.spark.sql.functions.col
>>>>
>>>> object SparkClientConnectionTest {
>>>>
>>>> private val CACHE_NAME = "SparkCache"
>>>>
>>>> private val CONFIG = "/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
>>>>
>>>> def setupExampleData = {
>>>>
>>>> val cfg2 = new ClientConfiguration().setAddresses("3.88.248.113:10800")
>>>> val igniteClient:IgniteClient = Ignition.startClient(cfg2)
>>>>
>>>> System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
>>>>
>>>> val cache:ClientCache[Integer, Address] = igniteClient.getOrCreateCache(CACHE_NAME)
>>>>
>>>> cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
>>>> .setSchema("PUBLIC")).getAll
>>>>
>>>> cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) ) WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
>>>> .setSchema("PUBLIC")).getAll
>>>>
>>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco", "04074").setSchema("PUBLIC")).getAll
>>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road", "520003").setSchema("PUBLIC")).getAll
>>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road", "1234").setSchema("PUBLIC")).getAll
>>>>
>>>> System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
>>>>
>>>> val data=cache.query(new SqlFieldsQuery("select * from Person").setSchema("PUBLIC")).getAll
>>>>
>>>> println(data.toString)
>>>> }
>>>>
>>>> def sparkDSLExample(implicit spark: SparkSession): Unit = {
>>>> println("Querying using Spark DSL.")
>>>> println
>>>>
>>>>
>>>> val igniteDF = spark.read
>>>> .format(FORMAT_IGNITE) //Data source type.
>>>> .option(OPTION_TABLE, "person") //Table to read.
>>>> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
>>>> .load()
>>>> .filter(col("id") >= 2) //Filter clause.
>>>> .filter(col("name") like "%J%") //Another filter clause.
>>>>
>>>> println("Data frame schema:")
>>>>
>>>> igniteDF.printSchema() //Printing query schema to console.
>>>>
>>>> println("Data frame content:")
>>>>
>>>> igniteDF.show() //Printing query results to console.
>>>> }
>>>>
>>>>
>>>> def main(args: Array[String]): Unit = {
>>>>
>>>> setupExampleData
>>>>
>>>> //Creating spark session.
>>>> implicit val spark = SparkSession.builder()
>>>> .appName("Spark Ignite data sources example")
>>>> .master("local")
>>>> .config("spark.executor.instances", "2")
>>>> .getOrCreate()
>>>>
>>>> // Adjust the logger to exclude the logs of no interest.
>>>> Logger.getRootLogger.setLevel(Level.ERROR)
>>>> Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
>>>>
>>>> //sparkDSLExample
>>>>
>>>>
>>>> val df = spark.read
>>>> .format("jdbc")
>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>> .option("fetchsize",100)
>>>> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
>>>> .option("dbtable", "Person").load()
>>>>
>>>> df.printSchema()
>>>>
>>>> df.createOrReplaceTempView("test")
>>>>
>>>> spark.sql("select * from test where id=1").show(10)
>>>>
>>>> spark.sql("select 4,'blah',124232").show(10)
>>>>
>>>> import java.sql.DriverManager
>>>> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>>>>
>>>> import java.util.Properties
>>>> val connectionProperties = new Properties()
>>>>
>>>> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>>>>
>>>> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
>>>> "Person",connectionProperties)
>>>>
>>>> spark.read
>>>> .format("jdbc")
>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>> .option("fetchsize",100)
>>>> .option("dbtable", "Person").load().show(10,false)
>>>>
>>>> }
>>>>
>>>> }
>>>>
>>>>
>>>> example-default.xml:-
>>>>
>>>> <?xml version="1.0" encoding="UTF-8"?>
>>>>
>>>> <!--
>>>> Licensed to the Apache Software Foundation (ASF) under one or more
>>>> contributor license agreements. See the NOTICE file distributed with
>>>> this work for additional information regarding copyright ownership.
>>>> The ASF licenses this file to You under the Apache License, Version 2.0
>>>> (the "License"); you may not use this file except in compliance with
>>>> the License. You may obtain a copy of the License at
>>>>
>>>> http://www.apache.org/licenses/LICENSE-2.0
>>>>
>>>> Unless required by applicable law or agreed to in writing, software
>>>> distributed under the License is distributed on an "AS IS" BASIS,
>>>> WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
>>>> See the License for the specific language governing permissions and
>>>> limitations under the License.
>>>> -->
>>>>
>>>> <!--
>>>> Ignite configuration with all defaults and enabled p2p deployment and enabled events.
>>>> -->
>>>> <beans xmlns="http://www.springframework.org/schema/beans"
>>>> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
>>>> xmlns:util="http://www.springframework.org/schema/util"
>>>> xsi:schemaLocation="
>>>> http://www.springframework.org/schema/beans
>>>> http://www.springframework.org/schema/beans/spring-beans.xsd
>>>> http://www.springframework.org/schema/util
>>>> http://www.springframework.org/schema/util/spring-util.xsd">
>>>> <bean abstract="true" id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
>>>> <!-- Set to true to enable distributed class loading for examples, default is false. -->
>>>> <property name="peerClassLoadingEnabled" value="true"/>
>>>>
>>>> <!-- Enable task execution events for examples. -->
>>>> <property name="includeEventTypes">
>>>> <list>
>>>> <!--Task execution events-->
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
>>>>
>>>> <!--Cache events-->
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
>>>> </list>
>>>> </property>
>>>>
>>>> <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
>>>> <property name="discoverySpi">
>>>> <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
>>>> <property name="ipFinder">
>>>> <!--
>>>> Ignite provides several options for automatic discovery that can be used
>>>> instead os static IP based discovery. For information on all options refer
>>>> to our documentation: http://apacheignite.readme.io/docs/cluster-config
>>>> -->
>>>> <!-- Uncomment static IP finder to enable static-based discovery of initial nodes. -->
>>>> <!--<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
>>>> <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
>>>> <property name="addresses">
>>>> <list>
>>>> <!-- In distributed environment, replace with actual host IP address. -->
>>>> <value>3.88.248.113:10800</value>
>>>> </list>
>>>> </property>
>>>> </bean>
>>>> </property>
>>>> </bean>
>>>> </property>
>>>> </bean>
>>>> </beans>
>>>>
>>>>
>>>>
>>>>
>>>> --
>>>> Thanks & Regards
>>>> Sri Tummala
>>>>
>>>>
>>>>
>>>>
>>>
>>> --
>>> Thanks & Regards
>>> Sri Tummala
>>>
>>>
>>
>> --
>> Thanks & Regards
>> Sri Tummala
>>
>>
>
> --
> Thanks & Regards
> Sri Tummala
>
>
--
Thanks & Regards
Sri Tummala
Re: Apache Spark + Ignite Connection Issue
Posted by sri hari kali charan Tummala <ka...@gmail.com>.
I got it so I have to build a jar move to aws ignite cluster node and run
it on the node.
Thanks
Sri
On Thursday, October 24, 2019, Denis Magda <dm...@apache.org> wrote:
> Sri,
>
> That's wrong to assume that the Ignite thin client is the only way to
> access data stored in AWS or other cloud environments. The problems you are
> experiencing are caused by the fact that your application is running on the
> laptop while the cluster (server nodes) is within an AWS network.
>
> If you use a thick/regular client, then servers running on AWS need to
> have a way to *open* network connections to the application. This won't
> happen if the laptop or dev environment doesn't have public IPs reachable
> from AWS. Plus, additional network configuration might be needed.
>
> Overall, thick clients work with AWS, but either an application needs to
> be deployed to AWS during development, or you need to start a local cluster
> in your network/premises first, accomplish development and move on with
> testing in AWS. A network configuration is also an option but a tricky one
> and might not be doable.
>
> Thin clients are a solution as well if key-value and SQL APIs are all you
> need.
>
> -
> Denis
>
>
> On Sun, Oct 20, 2019 at 2:06 PM sri hari kali charan Tummala <
> kali.tummala@gmail.com> wrote:
>
> so I got an answer from Grid Grain computing, JDBC thin client is the only
> way yo connect the ignite cluster which is running on aws.
>
> https://forums.gridgain.com/community-home/digestviewer/
> viewthread?MessageKey=13f5e836-1569-486a-8475-84c70fc141e0&CommunityKey=
> 3b551477-7e2d-462f-bc5f-d6d10ccbbe35&tab=digestviewer&
> SuccessMsg=Thank+you+for+submitting+your+message.
>
> Thanks
> Sri
>
> On Fri, Oct 18, 2019 at 1:38 PM sri hari kali charan Tummala <
> kali.tummala@gmail.com> wrote:
>
> Ok I created a 2 node ec2 instance ignite cluster is below is the right
> way to create cache? the code still using my laptop resources unable to
> connect ignite cluster.
>
> Out put:-
> [13:36:36] Ignite node started OK (id=8535f4d3)
> [13:36:36] Topology snapshot [ver=30, locNode=8535f4d3, *servers=1,*
> clients=1, state=ACTIVE, CPUs=8, *offheap=6.4GB, heap=14.0GB*]
> >>> cache acquired
>
> package com.ignite.examples.igniteStartup
>
> import java.util.Arrays
> import java.util.List
> import com.ignite.examples.model.Address
> import org.apache.commons.logging.Log
> import org.apache.commons.logging.LogFactory
> import org.apache.ignite.Ignite
> import org.apache.ignite.IgniteCache
> import org.apache.ignite.Ignition
> import org.apache.ignite.configuration.IgniteConfiguration
> import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi
> import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder
>
> //remove if not needed
> import scala.collection.JavaConversions._
>
> object IgniteStart2 {
>
> private var cache: IgniteCache[String, Address] = _
>
> def main(args: Array[String]): Unit = {
>
> val spi: TcpDiscoverySpi = new TcpDiscoverySpi()
> val ipFinder: TcpDiscoveryVmIpFinder = new TcpDiscoveryVmIpFinder()
> val hostList: List[String] = Arrays.asList("127.0.0.1:47500..47509,18.206.247.40:47500..47509,52.207.217.31:47500..47509".split(","): _*)
> ipFinder.setAddresses(hostList)
> spi.setIpFinder(ipFinder)
> val cfg: IgniteConfiguration = new IgniteConfiguration()
> cfg.setDiscoverySpi(spi)
> cfg.setClientMode(true)
> System.out.println(">>> I am here")
>
> cfg.setPeerClassLoadingEnabled(true)
> val ignite: Ignite = Ignition.start(cfg)
> cache = Ignition.ignite().cache("test")
> System.out.println(">>> cache acquired")
>
> System.exit(0)
>
>
> }
>
> }
>
>
> On Fri, Oct 18, 2019 at 12:56 PM sri hari kali charan Tummala <
> kali.tummala@gmail.com> wrote:
>
> Hi Stephen ,
>
> I followed below steps and created one node ec2 instance with Ignite on it
> now I am connecting to the ignite cluster using spark in two ways 1) thin
> client 2) I don't know you have to tell me
>
> my question is what is the value I have to pass for option 2 is it ec2
> instance public IP with port 47500..47509 in my example-default.xml file ?
>
> https://www.gridgain.com/docs/8.7.6//installation-guide/
> manual-install-on-ec2
>
> Thanks
> Sri
>
> On Fri, Oct 18, 2019 at 12:47 PM Stephen Darlington <
> stephen.darlington@gridgain.com> wrote:
>
> You’re still pointing your Spark node (*thick* client) at port 10800 (the
> *thin* client port). This is not going to work.
>
> You can create a table using sqlline and read it using Spark, or vice
> versa. But you need a functioning cluster.
>
> Check out the documentation on clustering concepts and configuration:
> https://www.gridgain.com/docs/latest/developers-guide/clustering/
> clustering
>
> On 18 Oct 2019, at 16:32, sri hari kali charan Tummala <
> kali.tummala@gmail.com> wrote:
>
> Hi Stephen/All,
>
> got it working somewhat using below, but have an issue table and data
> which is created using thin client is failing to read using spark but table
> created by spark can be read using a thin client, that means table created
> in Ignite using spark are the only ones read using spark in Ignite?
>
> example-default.xml
>
> <property name="addresses">
> <list>
> <!-- In distributed environment, replace with actual host IP address. -->
> <value>18.206.247.40:10800</value>
> </list>
> </property>
>
>
> NotWorking Scala Function:-
>
> def readThinClientTableUsingSpark(implicit spark: SparkSession) = {
> val personDataFrame = spark.read
> .format(FORMAT_IGNITE)
> .option(OPTION_CONFIG_FILE, CONFIG)
> .option(OPTION_TABLE, "person")
> .load()
>
> println()
> println("Data frame thin connection content:person")
> println()
>
> //Printing content of data frame to console.
> personDataFrame.show()
> }
>
>
>
>
>
>
>
>
>
>
>
>
>
> *Exception in thread "main" class org.apache.ignite.IgniteException:
> Unknown table person at
> org.apache.ignite.spark.impl.IgniteSQLRelation$$anonfun$schema$2.apply(IgniteSQLRelation.scala:46)
> at
> org.apache.ignite.spark.impl.IgniteSQLRelation$$anonfun$schema$2.apply(IgniteSQLRelation.scala:46)
> at scala.Option.getOrElse(Option.scala:121) at
> org.apache.ignite.spark.impl.IgniteSQLRelation.schema(IgniteSQLRelation.scala:46)
> at
> org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:431)
> at
> org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:239)
> at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:227) at
> org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:164) at
> com.ignite.examples.spark.SparkIgniteCleanCode$.readThinClientTableUsingSpark(SparkIgniteCleanCode.scala:154)
> at
> com.ignite.examples.spark.SparkIgniteCleanCode$.main(SparkIgniteCleanCode.scala:216)
> at
> com.ignite.examples.spark.SparkIgniteCleanCode.main(SparkIgniteCleanCode.scala)*
>
> *Full Code:- (step 7 Fails)*
>
> package com.ignite.examples.spark
>
> import com.ignite.examples.model.Address
> import org.apache.ignite.{Ignite, Ignition}
> import org.apache.ignite.cache.query.SqlFieldsQuery
> import org.apache.ignite.client.{ClientCache, IgniteClient}
> import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
> import java.lang.{Long => JLong, String => JString}
>
> import com.ignite.examples.spark.SparkClientConnectionTest.{CACHE_NAME, CONFIG}
> import org.apache.ignite.internal.util.IgniteUtils.resolveIgnitePath
> import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE, OPTION_CONFIG_FILE, OPTION_CREATE_TABLE_PARAMETERS, OPTION_CREATE_TABLE_PRIMARY_KEY_FIELDS, OPTION_TABLE}
> import org.apache.log4j.{Level, Logger}
> import org.apache.spark.sql.{SaveMode, SparkSession}
>
> object SparkIgniteCleanCode {
>
> private val CACHE_NAME = "SparkCache"
>
> private val CONFIG = "/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
>
> def setupExampleData = {
>
> val cfg2 = new ClientConfiguration().setAddresses("18.206.247.40:10800")
> val igniteClient:IgniteClient = Ignition.startClient(cfg2)
>
> System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
>
> val cache:ClientCache[Integer, Address] = igniteClient.getOrCreateCache(CACHE_NAME)
>
> cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
> .setSchema("PUBLIC")).getAll
>
> cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) ) WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
> .setSchema("PUBLIC")).getAll
>
> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco", "04074").setSchema("PUBLIC")).getAll
> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road", "520003").setSchema("PUBLIC")).getAll
> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road", "1234").setSchema("PUBLIC")).getAll
>
> System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
>
> val data=cache.query(new SqlFieldsQuery("select * from Person").setSchema("PUBLIC")).getAll
>
> println(data.toString)
> }
>
> def sparkReadIgniteWithThinClient(implicit spark: SparkSession)={
>
> val df = spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://18.206.247.40")
> .option("fetchsize",100)
> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
> .option("dbtable", "Person").load()
>
> df.printSchema()
>
> df.createOrReplaceTempView("test")
>
> spark.sql("select * from test where id=1").show(10)
>
> spark.sql("select 4,'blah',124232").show(10)
>
> }
>
> def sparkWriteIgniteWithThinClient(implicit spark: SparkSession)={
>
> import java.sql.DriverManager
> val connection = DriverManager.getConnection("jdbc:ignite:thin://18.206.247.40")
>
> import java.util.Properties
> val connectionProperties = new Properties()
>
> connectionProperties.put("url", "jdbc:ignite:thin://18.206.247.40")
>
> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://18.206.247.40",
> "Person",connectionProperties)
>
> spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://18.206.247.40")
> .option("fetchsize",100)
> .option("dbtable", "Person").load().show(10,false)
>
> }
>
> def writeJSonToIgniteUsingSpark(implicit spark: SparkSession): Unit = {
>
> val ignite = Ignition.start(CONFIG)
>
> //Load content of json file to data frame.
> val personsDataFrame = spark.read.json(
> resolveIgnitePath("/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/resources/person.json").getAbsolutePath)
>
> println()
> println("Json file content:")
> println()
>
> //Printing content of json file to console.
> personsDataFrame.show()
>
> println()
> println("Writing Data Frame to Ignite:")
> println()
>
> //Writing content of data frame to Ignite.
> personsDataFrame.write
> .format(FORMAT_IGNITE)
> .mode(org.apache.spark.sql.SaveMode.Append)
> .option(OPTION_CONFIG_FILE, CONFIG)
> .option(OPTION_TABLE, "json_person")
> .option(OPTION_CREATE_TABLE_PRIMARY_KEY_FIELDS, "id")
> .option(OPTION_CREATE_TABLE_PARAMETERS, "template=replicated")
> .save()
>
> println("Done!")
>
> println()
> println("Reading data from Ignite table:")
> println()
>
> val cache = ignite.cache[Any, Any](CACHE_NAME)
>
> //Reading saved data from Ignite.
> val data = cache.query(new SqlFieldsQuery("SELECT id, name, department FROM json_person")).getAll
>
> println(data.toString)
>
> //data.foreach { row ⇒ println(row.mkString("[", ", ", "]")) }
> }
>
> def readIgniteUsingSpark(implicit spark: SparkSession) = {
> val json_person = spark.read
> .format(FORMAT_IGNITE)
> .option(OPTION_CONFIG_FILE, CONFIG)
> .option(OPTION_TABLE, "json_person")
> .load()
>
> println()
> println("Data frame content:json_person")
> println()
>
> //Printing content of data frame to console.
> json_person.show()
> }
>
>
> def readThinClientTableUsingSpark(implicit spark: SparkSession) = {
> val personDataFrame = spark.read
> .format(FORMAT_IGNITE)
> .option(OPTION_CONFIG_FILE, CONFIG)
> .option(OPTION_TABLE, "person")
> .load()
>
> println()
> println("Data frame thin connection content:person")
> println()
>
> //Printing content of data frame to console.
> personDataFrame.show()
> }
>
>
> def main(args: Array[String]): Unit = {
>
> println()
> println("Step 1 setupExampleData:")
> println()
>
> setupExampleData
>
> println()
> println("Step 2 createSparkSession:")
> println()
>
> //Creating spark session.
> implicit val spark = SparkSession.builder()
> .appName("Spark Ignite data sources example")
> .master("local")
> .config("spark.executor.instances", "2")
> .getOrCreate()
>
> // Adjust the logger to exclude the logs of no interest.
> Logger.getRootLogger.setLevel(Level.ERROR)
> Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
>
> println()
> println("Step 3 ReadIgniteWithThinClient of Step1 Data:")
> println()
>
> sparkReadIgniteWithThinClient(spark)
>
> println()
> println("Step 4 sparkWriteIgniteWithThinClient of Step1 Data:")
> println()
>
> sparkWriteIgniteWithThinClient(spark)
>
> println()
> println("Step 5 writeJSonToIgniteUsingSpark Using Spark:")
> println()
>
> writeJSonToIgniteUsingSpark(spark)
>
> println()
> println("Step 6 readIgniteUsingSpark Using Spark:")
> println()
>
> readIgniteUsingSpark(spark)
>
> println()
> println("Step 7 readThinClientTableUsingSpark Using Spark:")
> println()
>
> readThinClientTableUsingSpark(spark)
>
>
> spark.stop()
>
>
> }
>
> }
>
>
> example-default.xml
>
> <?xml version="1.0" encoding="UTF-8"?>
>
> <!--
> Licensed to the Apache Software Foundation (ASF) under one or more
> contributor license agreements. See the NOTICE file distributed with
> this work for additional information regarding copyright ownership.
> The ASF licenses this file to You under the Apache License, Version 2.0
> (the "License"); you may not use this file except in compliance with
> the License. You may obtain a copy of the License at
>
> http://www.apache.org/licenses/LICENSE-2.0
>
> Unless required by applicable law or agreed to in writing, software
> distributed under the License is distributed on an "AS IS" BASIS,
> WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
> See the License for the specific language governing permissions and
> limitations under the License.
> -->
>
> <!--
> Ignite configuration with all defaults and enabled p2p deployment and enabled events.
> -->
> <beans xmlns="http://www.springframework.org/schema/beans"
> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
> xmlns:util="http://www.springframework.org/schema/util"
> xsi:schemaLocation="
> http://www.springframework.org/schema/beans
> http://www.springframework.org/schema/beans/spring-beans.xsd
> http://www.springframework.org/schema/util
> http://www.springframework.org/schema/util/spring-util.xsd">
> <bean abstract="true" id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
> <!-- Set to true to enable distributed class loading for examples, default is false. -->
> <property name="peerClassLoadingEnabled" value="true"/>
>
> <!-- Enable task execution events for examples. -->
> <property name="includeEventTypes">
> <list>
> <!--Task execution events-->
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
>
> <!--Cache events-->
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
> </list>
> </property>
>
> <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
> <property name="discoverySpi">
> <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
> <property name="ipFinder">
> <!--
> Ignite provides several options for automatic discovery that can be used
> instead os static IP based discovery. For information on all options refer
> to our documentation: http://apacheignite.readme.io/docs/cluster-config
> -->
> <!-- Uncomment static IP finder to enable static-based discovery of initial nodes. -->
> <!--<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
> <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
> <property name="addresses">
> <list>
> <!-- In distributed environment, replace with actual host IP address. -->
> <value>18.206.247.40:10800</value>
> </list>
> </property>
> </bean>
> </property>
> </bean>
> </property>
> </bean>
> </beans>
>
>
> Thanks
> Sri
>
> On Fri, Oct 18, 2019 at 9:23 AM sri hari kali charan Tummala <
> kali.tummala@gmail.com> wrote:
>
> do you mean communication ports 47100-47200 as mentioned (
> https://www.gridgain.com/docs/8.7.6//installation-
> guide/manual-install-on-ec2) ? bare in mind I am running my spark job
> outside ignite ec2 box (MAC PC).
>
> which option is right in my default.xml file?
>
> Option 1:-
>
> <property name="addresses">
> <list>
> <!-- In distributed environment, replace with actual host IP address. -->
> <value>3.88.248.113:4 <http://3.88.248.113:10800/>7100..47200</value>
> </list>
> </property>
>
>
> Option 2:-
>
> <property name="addresses">
> <list>
> <!-- In distributed environment, replace with actual host IP address. -->
> <value>3.88.248.113:4 <http://3.88.248.113:10800/>7500..47600</value>
> </list>
> </property>
>
>
> Option 3:-
>
> <property name="addresses">
> <list>
> <!-- In distributed environment, replace with actual host IP address. -->
> <value>3.88.248.113 <http://3.88.248.113:10800/></value>
> </list>
> </property>
>
>
>
> Thanks
>
> On Fri, Oct 18, 2019 at 9:18 AM sri hari kali charan Tummala <
> kali.tummala@gmail.com> wrote:
>
> Hi Stephen ,
>
> do you mean 3.88.248.113: <http://3.88.248.113:10800/>47500..47700
> something like this? or just public ip 3.88.248.113
> <http://3.88.248.113:10800/> I tried all the possibilities none of them
> are getting connected.
>
> Thanks
> Sri
>
> On Fri, Oct 18, 2019 at 6:02 AM Stephen Darlington <
> stephen.darlington@gridgain.com> wrote:
>
> You’re trying to connect a thick client (the Spark integration) to the
> thin client port (10800). Your example-default.xml file needs to have the
> same configuration as your server node(s).
>
> Regards,
> Stephen
>
> On 17 Oct 2019, at 18:12, sri hari kali charan Tummala <
> kali.tummala@gmail.com> wrote:
>
> Hi Community,
>
> I am trying to read and write into the Ignite cluster using apache-spark I
> am able to do that using JDBC thin client but not native method as
> mentioned in several spark + ignite examples.
>
> Right now all the spark + ignite examples launch a local ignite cluster
> but I want my code connecting to already existing cluster (client).
>
> Question:-
> *How to pass Ignite connection ip and port (10800) 10800 in
> example-default.xml ?*
>
> Error:-
> *TcpDiscoverySpi: Failed to connect to any address from IP finder (will
> retry to join topology every 2000 ms; change 'reconnectDelay' to configure
> the frequency of retries): [/3.88.248.113:10800
> <http://3.88.248.113:10800/>]*
>
> Working (Spark + Ignite using JDBC):-
>
> val df = spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://3.88.248.113")
> .option("fetchsize",100)
> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
> .option("dbtable", "Person").load()
>
> df.printSchema()
>
> df.createOrReplaceTempView("test")
>
> spark.sql("select * from test where id=1").show(10)
>
> spark.sql("select 4,'blah',124232").show(10)
>
> import java.sql.DriverManager
> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>
> import java.util.Properties
> val connectionProperties = new Properties()
>
> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>
> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
> "Person",connectionProperties)
>
> spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://3.88.248.113")
> .option("fetchsize",100)
> .option("dbtable", "Person").load().show(10,false)
>
>
> Not Working requires a CONFIG file which is example-default.xml:-
>
> val igniteDF = spark.read
> .format(FORMAT_IGNITE) //Data source type.
> .option(OPTION_TABLE, "person") //Table to read.
> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
> .load()
> .filter(col("id") >= 2) //Filter clause.
> .filter(col("name") like "%J%") //Another filter clause.
>
>
> Full Code:- (sparkDSLExample) function fails to connect ignite cluster
> which I already have
>
> package com.ignite.examples.spark
>
> import com.ignite.examples.model.Address
> import org.apache.ignite.{Ignite, Ignition}
> import org.apache.ignite.cache.query.SqlFieldsQuery
> import org.apache.ignite.client.{ClientCache, IgniteClient}
> import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
> import java.lang.{Long => JLong, String => JString}
>
> import org.apache.ignite.cache.query.SqlFieldsQuery
> import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE, OPTION_CONFIG_FILE, OPTION_TABLE}
> import org.apache.log4j.{Level, Logger}
> import org.apache.spark.sql.{SaveMode, SparkSession}
> import org.apache.spark.sql.functions.col
>
> object SparkClientConnectionTest {
>
> private val CACHE_NAME = "SparkCache"
>
> private val CONFIG = "/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
>
> def setupExampleData = {
>
> val cfg2 = new ClientConfiguration().setAddresses("3.88.248.113:10800")
> val igniteClient:IgniteClient = Ignition.startClient(cfg2)
>
> System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
>
> val cache:ClientCache[Integer, Address] = igniteClient.getOrCreateCache(CACHE_NAME)
>
> cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
> .setSchema("PUBLIC")).getAll
>
> cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) ) WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
> .setSchema("PUBLIC")).getAll
>
> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco", "04074").setSchema("PUBLIC")).getAll
> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road", "520003").setSchema("PUBLIC")).getAll
> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road", "1234").setSchema("PUBLIC")).getAll
>
> System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
>
> val data=cache.query(new SqlFieldsQuery("select * from Person").setSchema("PUBLIC")).getAll
>
> println(data.toString)
> }
>
> def sparkDSLExample(implicit spark: SparkSession): Unit = {
> println("Querying using Spark DSL.")
> println
>
>
> val igniteDF = spark.read
> .format(FORMAT_IGNITE) //Data source type.
> .option(OPTION_TABLE, "person") //Table to read.
> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
> .load()
> .filter(col("id") >= 2) //Filter clause.
> .filter(col("name") like "%J%") //Another filter clause.
>
> println("Data frame schema:")
>
> igniteDF.printSchema() //Printing query schema to console.
>
> println("Data frame content:")
>
> igniteDF.show() //Printing query results to console.
> }
>
>
> def main(args: Array[String]): Unit = {
>
> setupExampleData
>
> //Creating spark session.
> implicit val spark = SparkSession.builder()
> .appName("Spark Ignite data sources example")
> .master("local")
> .config("spark.executor.instances", "2")
> .getOrCreate()
>
> // Adjust the logger to exclude the logs of no interest.
> Logger.getRootLogger.setLevel(Level.ERROR)
> Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
>
> //sparkDSLExample
>
>
> val df = spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://3.88.248.113")
> .option("fetchsize",100)
> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
> .option("dbtable", "Person").load()
>
> df.printSchema()
>
> df.createOrReplaceTempView("test")
>
> spark.sql("select * from test where id=1").show(10)
>
> spark.sql("select 4,'blah',124232").show(10)
>
> import java.sql.DriverManager
> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>
> import java.util.Properties
> val connectionProperties = new Properties()
>
> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>
> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
> "Person",connectionProperties)
>
> spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://3.88.248.113")
> .option("fetchsize",100)
> .option("dbtable", "Person").load().show(10,false)
>
> }
>
> }
>
>
> example-default.xml:-
>
> <?xml version="1.0" encoding="UTF-8"?>
>
> <!--
> Licensed to the Apache Software Foundation (ASF) under one or more
> contributor license agreements. See the NOTICE file distributed with
> this work for additional information regarding copyright ownership.
> The ASF licenses this file to You under the Apache License, Version 2.0
> (the "License"); you may not use this file except in compliance with
> the License. You may obtain a copy of the License at
>
> http://www.apache.org/licenses/LICENSE-2.0
>
> Unless required by applicable law or agreed to in writing, software
> distributed under the License is distributed on an "AS IS" BASIS,
> WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
> See the License for the specific language governing permissions and
> limitations under the License.
> -->
>
> <!--
> Ignite configuration with all defaults and enabled p2p deployment and enabled events.
> -->
> <beans xmlns="http://www.springframework.org/schema/beans"
> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
> xmlns:util="http://www.springframework.org/schema/util"
> xsi:schemaLocation="
> http://www.springframework.org/schema/beans
> http://www.springframework.org/schema/beans/spring-beans.xsd
> http://www.springframework.org/schema/util
> http://www.springframework.org/schema/util/spring-util.xsd">
> <bean abstract="true" id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
> <!-- Set to true to enable distributed class loading for examples, default is false. -->
> <property name="peerClassLoadingEnabled" value="true"/>
>
> <!-- Enable task execution events for examples. -->
> <property name="includeEventTypes">
> <list>
> <!--Task execution events-->
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
>
> <!--Cache events-->
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
> </list>
> </property>
>
> <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
> <property name="discoverySpi">
> <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
> <property name="ipFinder">
> <!--
> Ignite provides several options for automatic discovery that can be used
> instead os static IP based discovery. For information on all options refer
> to our documentation: http://apacheignite.readme.io/docs/cluster-config
> -->
> <!-- Uncomment static IP finder to enable static-based discovery of initial nodes. -->
> <!--<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
> <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
> <property name="addresses">
> <list>
> <!-- In distributed environment, replace with actual host IP address. -->
> <value>3.88.248.113:10800</value>
>
>
--
Thanks & Regards
Sri Tummala
Re: Apache Spark + Ignite Connection Issue
Posted by Denis Magda <dm...@apache.org>.
Sri,
That's wrong to assume that the Ignite thin client is the only way to
access data stored in AWS or other cloud environments. The problems you are
experiencing are caused by the fact that your application is running on the
laptop while the cluster (server nodes) is within an AWS network.
If you use a thick/regular client, then servers running on AWS need to have
a way to *open* network connections to the application. This won't happen
if the laptop or dev environment doesn't have public IPs reachable from
AWS. Plus, additional network configuration might be needed.
Overall, thick clients work with AWS, but either an application needs to be
deployed to AWS during development, or you need to start a local cluster in
your network/premises first, accomplish development and move on with
testing in AWS. A network configuration is also an option but a tricky one
and might not be doable.
Thin clients are a solution as well if key-value and SQL APIs are all you
need.
-
Denis
On Sun, Oct 20, 2019 at 2:06 PM sri hari kali charan Tummala <
kali.tummala@gmail.com> wrote:
> so I got an answer from Grid Grain computing, JDBC thin client is the only
> way yo connect the ignite cluster which is running on aws.
>
>
> https://forums.gridgain.com/community-home/digestviewer/viewthread?MessageKey=13f5e836-1569-486a-8475-84c70fc141e0&CommunityKey=3b551477-7e2d-462f-bc5f-d6d10ccbbe35&tab=digestviewer&SuccessMsg=Thank+you+for+submitting+your+message.
>
> Thanks
> Sri
>
> On Fri, Oct 18, 2019 at 1:38 PM sri hari kali charan Tummala <
> kali.tummala@gmail.com> wrote:
>
>> Ok I created a 2 node ec2 instance ignite cluster is below is the right
>> way to create cache? the code still using my laptop resources unable to
>> connect ignite cluster.
>>
>> Out put:-
>> [13:36:36] Ignite node started OK (id=8535f4d3)
>> [13:36:36] Topology snapshot [ver=30, locNode=8535f4d3, *servers=1,*
>> clients=1, state=ACTIVE, CPUs=8, *offheap=6.4GB, heap=14.0GB*]
>> >>> cache acquired
>>
>> package com.ignite.examples.igniteStartup
>>
>> import java.util.Arrays
>> import java.util.List
>> import com.ignite.examples.model.Address
>> import org.apache.commons.logging.Log
>> import org.apache.commons.logging.LogFactory
>> import org.apache.ignite.Ignite
>> import org.apache.ignite.IgniteCache
>> import org.apache.ignite.Ignition
>> import org.apache.ignite.configuration.IgniteConfiguration
>> import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi
>> import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder
>>
>> //remove if not needed
>> import scala.collection.JavaConversions._
>>
>> object IgniteStart2 {
>>
>> private var cache: IgniteCache[String, Address] = _
>>
>> def main(args: Array[String]): Unit = {
>>
>> val spi: TcpDiscoverySpi = new TcpDiscoverySpi()
>> val ipFinder: TcpDiscoveryVmIpFinder = new TcpDiscoveryVmIpFinder()
>> val hostList: List[String] = Arrays.asList("127.0.0.1:47500..47509,18.206.247.40:47500..47509,52.207.217.31:47500..47509".split(","): _*)
>> ipFinder.setAddresses(hostList)
>> spi.setIpFinder(ipFinder)
>> val cfg: IgniteConfiguration = new IgniteConfiguration()
>> cfg.setDiscoverySpi(spi)
>> cfg.setClientMode(true)
>> System.out.println(">>> I am here")
>>
>> cfg.setPeerClassLoadingEnabled(true)
>> val ignite: Ignite = Ignition.start(cfg)
>> cache = Ignition.ignite().cache("test")
>> System.out.println(">>> cache acquired")
>>
>> System.exit(0)
>>
>>
>> }
>>
>> }
>>
>>
>> On Fri, Oct 18, 2019 at 12:56 PM sri hari kali charan Tummala <
>> kali.tummala@gmail.com> wrote:
>>
>>> Hi Stephen ,
>>>
>>> I followed below steps and created one node ec2 instance with Ignite on
>>> it now I am connecting to the ignite cluster using spark in two ways 1)
>>> thin client 2) I don't know you have to tell me
>>>
>>> my question is what is the value I have to pass for option 2 is it ec2
>>> instance public IP with port 47500..47509 in my example-default.xml file ?
>>>
>>>
>>> https://www.gridgain.com/docs/8.7.6//installation-guide/manual-install-on-ec2
>>>
>>> Thanks
>>> Sri
>>>
>>> On Fri, Oct 18, 2019 at 12:47 PM Stephen Darlington <
>>> stephen.darlington@gridgain.com> wrote:
>>>
>>>> You’re still pointing your Spark node (*thick* client) at port 10800
>>>> (the *thin* client port). This is not going to work.
>>>>
>>>> You can create a table using sqlline and read it using Spark, or vice
>>>> versa. But you need a functioning cluster.
>>>>
>>>> Check out the documentation on clustering concepts and configuration:
>>>> https://www.gridgain.com/docs/latest/developers-guide/clustering/clustering
>>>>
>>>>
>>>> On 18 Oct 2019, at 16:32, sri hari kali charan Tummala <
>>>> kali.tummala@gmail.com> wrote:
>>>>
>>>> Hi Stephen/All,
>>>>
>>>> got it working somewhat using below, but have an issue table and data
>>>> which is created using thin client is failing to read using spark but table
>>>> created by spark can be read using a thin client, that means table created
>>>> in Ignite using spark are the only ones read using spark in Ignite?
>>>>
>>>> example-default.xml
>>>>
>>>> <property name="addresses">
>>>> <list>
>>>> <!-- In distributed environment, replace with actual host IP address. -->
>>>> <value>18.206.247.40:10800</value>
>>>> </list>
>>>> </property>
>>>>
>>>>
>>>> NotWorking Scala Function:-
>>>>
>>>> def readThinClientTableUsingSpark(implicit spark: SparkSession) = {
>>>> val personDataFrame = spark.read
>>>> .format(FORMAT_IGNITE)
>>>> .option(OPTION_CONFIG_FILE, CONFIG)
>>>> .option(OPTION_TABLE, "person")
>>>> .load()
>>>>
>>>> println()
>>>> println("Data frame thin connection content:person")
>>>> println()
>>>>
>>>> //Printing content of data frame to console.
>>>> personDataFrame.show()
>>>> }
>>>>
>>>>
>>>>
>>>>
>>>>
>>>>
>>>>
>>>>
>>>>
>>>>
>>>>
>>>>
>>>>
>>>> *Exception in thread "main" class org.apache.ignite.IgniteException:
>>>> Unknown table person at
>>>> org.apache.ignite.spark.impl.IgniteSQLRelation$$anonfun$schema$2.apply(IgniteSQLRelation.scala:46)
>>>> at
>>>> org.apache.ignite.spark.impl.IgniteSQLRelation$$anonfun$schema$2.apply(IgniteSQLRelation.scala:46)
>>>> at scala.Option.getOrElse(Option.scala:121) at
>>>> org.apache.ignite.spark.impl.IgniteSQLRelation.schema(IgniteSQLRelation.scala:46)
>>>> at
>>>> org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:431)
>>>> at
>>>> org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:239)
>>>> at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:227) at
>>>> org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:164) at
>>>> com.ignite.examples.spark.SparkIgniteCleanCode$.readThinClientTableUsingSpark(SparkIgniteCleanCode.scala:154)
>>>> at
>>>> com.ignite.examples.spark.SparkIgniteCleanCode$.main(SparkIgniteCleanCode.scala:216)
>>>> at
>>>> com.ignite.examples.spark.SparkIgniteCleanCode.main(SparkIgniteCleanCode.scala)*
>>>>
>>>> *Full Code:- (step 7 Fails)*
>>>>
>>>> package com.ignite.examples.spark
>>>>
>>>> import com.ignite.examples.model.Address
>>>> import org.apache.ignite.{Ignite, Ignition}
>>>> import org.apache.ignite.cache.query.SqlFieldsQuery
>>>> import org.apache.ignite.client.{ClientCache, IgniteClient}
>>>> import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
>>>> import java.lang.{Long => JLong, String => JString}
>>>>
>>>> import com.ignite.examples.spark.SparkClientConnectionTest.{CACHE_NAME, CONFIG}
>>>> import org.apache.ignite.internal.util.IgniteUtils.resolveIgnitePath
>>>> import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE, OPTION_CONFIG_FILE, OPTION_CREATE_TABLE_PARAMETERS, OPTION_CREATE_TABLE_PRIMARY_KEY_FIELDS, OPTION_TABLE}
>>>> import org.apache.log4j.{Level, Logger}
>>>> import org.apache.spark.sql.{SaveMode, SparkSession}
>>>>
>>>> object SparkIgniteCleanCode {
>>>>
>>>> private val CACHE_NAME = "SparkCache"
>>>>
>>>> private val CONFIG = "/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
>>>>
>>>> def setupExampleData = {
>>>>
>>>> val cfg2 = new ClientConfiguration().setAddresses("18.206.247.40:10800")
>>>> val igniteClient:IgniteClient = Ignition.startClient(cfg2)
>>>>
>>>> System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
>>>>
>>>> val cache:ClientCache[Integer, Address] = igniteClient.getOrCreateCache(CACHE_NAME)
>>>>
>>>> cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
>>>> .setSchema("PUBLIC")).getAll
>>>>
>>>> cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) ) WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
>>>> .setSchema("PUBLIC")).getAll
>>>>
>>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco", "04074").setSchema("PUBLIC")).getAll
>>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road", "520003").setSchema("PUBLIC")).getAll
>>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road", "1234").setSchema("PUBLIC")).getAll
>>>>
>>>> System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
>>>>
>>>> val data=cache.query(new SqlFieldsQuery("select * from Person").setSchema("PUBLIC")).getAll
>>>>
>>>> println(data.toString)
>>>> }
>>>>
>>>> def sparkReadIgniteWithThinClient(implicit spark: SparkSession)={
>>>>
>>>> val df = spark.read
>>>> .format("jdbc")
>>>> .option("url", "jdbc:ignite:thin://18.206.247.40")
>>>> .option("fetchsize",100)
>>>> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
>>>> .option("dbtable", "Person").load()
>>>>
>>>> df.printSchema()
>>>>
>>>> df.createOrReplaceTempView("test")
>>>>
>>>> spark.sql("select * from test where id=1").show(10)
>>>>
>>>> spark.sql("select 4,'blah',124232").show(10)
>>>>
>>>> }
>>>>
>>>> def sparkWriteIgniteWithThinClient(implicit spark: SparkSession)={
>>>>
>>>> import java.sql.DriverManager
>>>> val connection = DriverManager.getConnection("jdbc:ignite:thin://18.206.247.40")
>>>>
>>>> import java.util.Properties
>>>> val connectionProperties = new Properties()
>>>>
>>>> connectionProperties.put("url", "jdbc:ignite:thin://18.206.247.40")
>>>>
>>>> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://18.206.247.40",
>>>> "Person",connectionProperties)
>>>>
>>>> spark.read
>>>> .format("jdbc")
>>>> .option("url", "jdbc:ignite:thin://18.206.247.40")
>>>> .option("fetchsize",100)
>>>> .option("dbtable", "Person").load().show(10,false)
>>>>
>>>> }
>>>>
>>>> def writeJSonToIgniteUsingSpark(implicit spark: SparkSession): Unit = {
>>>>
>>>> val ignite = Ignition.start(CONFIG)
>>>>
>>>> //Load content of json file to data frame.
>>>> val personsDataFrame = spark.read.json(
>>>> resolveIgnitePath("/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/resources/person.json").getAbsolutePath)
>>>>
>>>> println()
>>>> println("Json file content:")
>>>> println()
>>>>
>>>> //Printing content of json file to console.
>>>> personsDataFrame.show()
>>>>
>>>> println()
>>>> println("Writing Data Frame to Ignite:")
>>>> println()
>>>>
>>>> //Writing content of data frame to Ignite.
>>>> personsDataFrame.write
>>>> .format(FORMAT_IGNITE)
>>>> .mode(org.apache.spark.sql.SaveMode.Append)
>>>> .option(OPTION_CONFIG_FILE, CONFIG)
>>>> .option(OPTION_TABLE, "json_person")
>>>> .option(OPTION_CREATE_TABLE_PRIMARY_KEY_FIELDS, "id")
>>>> .option(OPTION_CREATE_TABLE_PARAMETERS, "template=replicated")
>>>> .save()
>>>>
>>>> println("Done!")
>>>>
>>>> println()
>>>> println("Reading data from Ignite table:")
>>>> println()
>>>>
>>>> val cache = ignite.cache[Any, Any](CACHE_NAME)
>>>>
>>>> //Reading saved data from Ignite.
>>>> val data = cache.query(new SqlFieldsQuery("SELECT id, name, department FROM json_person")).getAll
>>>>
>>>> println(data.toString)
>>>>
>>>> //data.foreach { row ⇒ println(row.mkString("[", ", ", "]")) }
>>>> }
>>>>
>>>> def readIgniteUsingSpark(implicit spark: SparkSession) = {
>>>> val json_person = spark.read
>>>> .format(FORMAT_IGNITE)
>>>> .option(OPTION_CONFIG_FILE, CONFIG)
>>>> .option(OPTION_TABLE, "json_person")
>>>> .load()
>>>>
>>>> println()
>>>> println("Data frame content:json_person")
>>>> println()
>>>>
>>>> //Printing content of data frame to console.
>>>> json_person.show()
>>>> }
>>>>
>>>>
>>>> def readThinClientTableUsingSpark(implicit spark: SparkSession) = {
>>>> val personDataFrame = spark.read
>>>> .format(FORMAT_IGNITE)
>>>> .option(OPTION_CONFIG_FILE, CONFIG)
>>>> .option(OPTION_TABLE, "person")
>>>> .load()
>>>>
>>>> println()
>>>> println("Data frame thin connection content:person")
>>>> println()
>>>>
>>>> //Printing content of data frame to console.
>>>> personDataFrame.show()
>>>> }
>>>>
>>>>
>>>> def main(args: Array[String]): Unit = {
>>>>
>>>> println()
>>>> println("Step 1 setupExampleData:")
>>>> println()
>>>>
>>>> setupExampleData
>>>>
>>>> println()
>>>> println("Step 2 createSparkSession:")
>>>> println()
>>>>
>>>> //Creating spark session.
>>>> implicit val spark = SparkSession.builder()
>>>> .appName("Spark Ignite data sources example")
>>>> .master("local")
>>>> .config("spark.executor.instances", "2")
>>>> .getOrCreate()
>>>>
>>>> // Adjust the logger to exclude the logs of no interest.
>>>> Logger.getRootLogger.setLevel(Level.ERROR)
>>>> Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
>>>>
>>>> println()
>>>> println("Step 3 ReadIgniteWithThinClient of Step1 Data:")
>>>> println()
>>>>
>>>> sparkReadIgniteWithThinClient(spark)
>>>>
>>>> println()
>>>> println("Step 4 sparkWriteIgniteWithThinClient of Step1 Data:")
>>>> println()
>>>>
>>>> sparkWriteIgniteWithThinClient(spark)
>>>>
>>>> println()
>>>> println("Step 5 writeJSonToIgniteUsingSpark Using Spark:")
>>>> println()
>>>>
>>>> writeJSonToIgniteUsingSpark(spark)
>>>>
>>>> println()
>>>> println("Step 6 readIgniteUsingSpark Using Spark:")
>>>> println()
>>>>
>>>> readIgniteUsingSpark(spark)
>>>>
>>>> println()
>>>> println("Step 7 readThinClientTableUsingSpark Using Spark:")
>>>> println()
>>>>
>>>> readThinClientTableUsingSpark(spark)
>>>>
>>>>
>>>> spark.stop()
>>>>
>>>>
>>>> }
>>>>
>>>> }
>>>>
>>>>
>>>> example-default.xml
>>>>
>>>> <?xml version="1.0" encoding="UTF-8"?>
>>>>
>>>> <!--
>>>> Licensed to the Apache Software Foundation (ASF) under one or more
>>>> contributor license agreements. See the NOTICE file distributed with
>>>> this work for additional information regarding copyright ownership.
>>>> The ASF licenses this file to You under the Apache License, Version 2.0
>>>> (the "License"); you may not use this file except in compliance with
>>>> the License. You may obtain a copy of the License at
>>>>
>>>> http://www.apache.org/licenses/LICENSE-2.0
>>>>
>>>> Unless required by applicable law or agreed to in writing, software
>>>> distributed under the License is distributed on an "AS IS" BASIS,
>>>> WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
>>>> See the License for the specific language governing permissions and
>>>> limitations under the License.
>>>> -->
>>>>
>>>> <!--
>>>> Ignite configuration with all defaults and enabled p2p deployment and enabled events.
>>>> -->
>>>> <beans xmlns="http://www.springframework.org/schema/beans"
>>>> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
>>>> xmlns:util="http://www.springframework.org/schema/util"
>>>> xsi:schemaLocation="
>>>> http://www.springframework.org/schema/beans
>>>> http://www.springframework.org/schema/beans/spring-beans.xsd
>>>> http://www.springframework.org/schema/util
>>>> http://www.springframework.org/schema/util/spring-util.xsd">
>>>> <bean abstract="true" id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
>>>> <!-- Set to true to enable distributed class loading for examples, default is false. -->
>>>> <property name="peerClassLoadingEnabled" value="true"/>
>>>>
>>>> <!-- Enable task execution events for examples. -->
>>>> <property name="includeEventTypes">
>>>> <list>
>>>> <!--Task execution events-->
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
>>>>
>>>> <!--Cache events-->
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
>>>> </list>
>>>> </property>
>>>>
>>>> <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
>>>> <property name="discoverySpi">
>>>> <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
>>>> <property name="ipFinder">
>>>> <!--
>>>> Ignite provides several options for automatic discovery that can be used
>>>> instead os static IP based discovery. For information on all options refer
>>>> to our documentation: http://apacheignite.readme.io/docs/cluster-config
>>>> -->
>>>> <!-- Uncomment static IP finder to enable static-based discovery of initial nodes. -->
>>>> <!--<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
>>>> <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
>>>> <property name="addresses">
>>>> <list>
>>>> <!-- In distributed environment, replace with actual host IP address. -->
>>>> <value>18.206.247.40:10800</value>
>>>> </list>
>>>> </property>
>>>> </bean>
>>>> </property>
>>>> </bean>
>>>> </property>
>>>> </bean>
>>>> </beans>
>>>>
>>>>
>>>> Thanks
>>>> Sri
>>>>
>>>> On Fri, Oct 18, 2019 at 9:23 AM sri hari kali charan Tummala <
>>>> kali.tummala@gmail.com> wrote:
>>>>
>>>>> do you mean communication ports 47100-47200 as mentioned (
>>>>> https://www.gridgain.com/docs/8.7.6//installation-guide/manual-install-on-ec2)
>>>>> ? bare in mind I am running my spark job outside ignite ec2 box (MAC PC).
>>>>>
>>>>> which option is right in my default.xml file?
>>>>>
>>>>> Option 1:-
>>>>>
>>>>> <property name="addresses">
>>>>> <list>
>>>>> <!-- In distributed environment, replace with actual host IP address. -->
>>>>> <value>3.88.248.113:4 <http://3.88.248.113:10800/>7100..47200</value>
>>>>> </list>
>>>>> </property>
>>>>>
>>>>>
>>>>> Option 2:-
>>>>>
>>>>> <property name="addresses">
>>>>> <list>
>>>>> <!-- In distributed environment, replace with actual host IP address. -->
>>>>> <value>3.88.248.113:4 <http://3.88.248.113:10800/>7500..47600</value>
>>>>> </list>
>>>>> </property>
>>>>>
>>>>>
>>>>> Option 3:-
>>>>>
>>>>> <property name="addresses">
>>>>> <list>
>>>>> <!-- In distributed environment, replace with actual host IP address. -->
>>>>> <value>3.88.248.113 <http://3.88.248.113:10800/></value>
>>>>> </list>
>>>>> </property>
>>>>>
>>>>>
>>>>>
>>>>> Thanks
>>>>>
>>>>> On Fri, Oct 18, 2019 at 9:18 AM sri hari kali charan Tummala <
>>>>> kali.tummala@gmail.com> wrote:
>>>>>
>>>>>> Hi Stephen ,
>>>>>>
>>>>>> do you mean 3.88.248.113: <http://3.88.248.113:10800/>47500..47700
>>>>>> something like this? or just public ip 3.88.248.113
>>>>>> <http://3.88.248.113:10800/> I tried all the possibilities none of
>>>>>> them are getting connected.
>>>>>>
>>>>>> Thanks
>>>>>> Sri
>>>>>>
>>>>>> On Fri, Oct 18, 2019 at 6:02 AM Stephen Darlington <
>>>>>> stephen.darlington@gridgain.com> wrote:
>>>>>>
>>>>>>> You’re trying to connect a thick client (the Spark integration) to
>>>>>>> the thin client port (10800). Your example-default.xml file needs to have
>>>>>>> the same configuration as your server node(s).
>>>>>>>
>>>>>>> Regards,
>>>>>>> Stephen
>>>>>>>
>>>>>>> On 17 Oct 2019, at 18:12, sri hari kali charan Tummala <
>>>>>>> kali.tummala@gmail.com> wrote:
>>>>>>>
>>>>>>> Hi Community,
>>>>>>>
>>>>>>> I am trying to read and write into the Ignite cluster using
>>>>>>> apache-spark I am able to do that using JDBC thin client but not native
>>>>>>> method as mentioned in several spark + ignite examples.
>>>>>>>
>>>>>>> Right now all the spark + ignite examples launch a local ignite
>>>>>>> cluster but I want my code connecting to already existing cluster (client).
>>>>>>>
>>>>>>> Question:-
>>>>>>> *How to pass Ignite connection ip and port (10800) 10800 in
>>>>>>> example-default.xml ?*
>>>>>>>
>>>>>>> Error:-
>>>>>>> *TcpDiscoverySpi: Failed to connect to any address from IP finder
>>>>>>> (will retry to join topology every 2000 ms; change 'reconnectDelay' to
>>>>>>> configure the frequency of retries): [/3.88.248.113:10800
>>>>>>> <http://3.88.248.113:10800/>]*
>>>>>>>
>>>>>>> Working (Spark + Ignite using JDBC):-
>>>>>>>
>>>>>>> val df = spark.read
>>>>>>> .format("jdbc")
>>>>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>>>>> .option("fetchsize",100)
>>>>>>> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
>>>>>>> .option("dbtable", "Person").load()
>>>>>>>
>>>>>>> df.printSchema()
>>>>>>>
>>>>>>> df.createOrReplaceTempView("test")
>>>>>>>
>>>>>>> spark.sql("select * from test where id=1").show(10)
>>>>>>>
>>>>>>> spark.sql("select 4,'blah',124232").show(10)
>>>>>>>
>>>>>>> import java.sql.DriverManager
>>>>>>> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>>>>>>>
>>>>>>> import java.util.Properties
>>>>>>> val connectionProperties = new Properties()
>>>>>>>
>>>>>>> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>>>>>>>
>>>>>>> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
>>>>>>> "Person",connectionProperties)
>>>>>>>
>>>>>>> spark.read
>>>>>>> .format("jdbc")
>>>>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>>>>> .option("fetchsize",100)
>>>>>>> .option("dbtable", "Person").load().show(10,false)
>>>>>>>
>>>>>>>
>>>>>>> Not Working requires a CONFIG file which is example-default.xml:-
>>>>>>>
>>>>>>> val igniteDF = spark.read
>>>>>>> .format(FORMAT_IGNITE) //Data source type.
>>>>>>> .option(OPTION_TABLE, "person") //Table to read.
>>>>>>> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
>>>>>>> .load()
>>>>>>> .filter(col("id") >= 2) //Filter clause.
>>>>>>> .filter(col("name") like "%J%") //Another filter clause.
>>>>>>>
>>>>>>>
>>>>>>> Full Code:- (sparkDSLExample) function fails to connect ignite
>>>>>>> cluster which I already have
>>>>>>>
>>>>>>> package com.ignite.examples.spark
>>>>>>>
>>>>>>> import com.ignite.examples.model.Address
>>>>>>> import org.apache.ignite.{Ignite, Ignition}
>>>>>>> import org.apache.ignite.cache.query.SqlFieldsQuery
>>>>>>> import org.apache.ignite.client.{ClientCache, IgniteClient}
>>>>>>> import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
>>>>>>> import java.lang.{Long => JLong, String => JString}
>>>>>>>
>>>>>>> import org.apache.ignite.cache.query.SqlFieldsQuery
>>>>>>> import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE, OPTION_CONFIG_FILE, OPTION_TABLE}
>>>>>>> import org.apache.log4j.{Level, Logger}
>>>>>>> import org.apache.spark.sql.{SaveMode, SparkSession}
>>>>>>> import org.apache.spark.sql.functions.col
>>>>>>>
>>>>>>> object SparkClientConnectionTest {
>>>>>>>
>>>>>>> private val CACHE_NAME = "SparkCache"
>>>>>>>
>>>>>>> private val CONFIG = "/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
>>>>>>>
>>>>>>> def setupExampleData = {
>>>>>>>
>>>>>>> val cfg2 = new ClientConfiguration().setAddresses("3.88.248.113:10800")
>>>>>>> val igniteClient:IgniteClient = Ignition.startClient(cfg2)
>>>>>>>
>>>>>>> System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
>>>>>>>
>>>>>>> val cache:ClientCache[Integer, Address] = igniteClient.getOrCreateCache(CACHE_NAME)
>>>>>>>
>>>>>>> cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
>>>>>>> .setSchema("PUBLIC")).getAll
>>>>>>>
>>>>>>> cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) ) WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
>>>>>>> .setSchema("PUBLIC")).getAll
>>>>>>>
>>>>>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco", "04074").setSchema("PUBLIC")).getAll
>>>>>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road", "520003").setSchema("PUBLIC")).getAll
>>>>>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road", "1234").setSchema("PUBLIC")).getAll
>>>>>>>
>>>>>>> System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
>>>>>>>
>>>>>>> val data=cache.query(new SqlFieldsQuery("select * from Person").setSchema("PUBLIC")).getAll
>>>>>>>
>>>>>>> println(data.toString)
>>>>>>> }
>>>>>>>
>>>>>>> def sparkDSLExample(implicit spark: SparkSession): Unit = {
>>>>>>> println("Querying using Spark DSL.")
>>>>>>> println
>>>>>>>
>>>>>>>
>>>>>>> val igniteDF = spark.read
>>>>>>> .format(FORMAT_IGNITE) //Data source type.
>>>>>>> .option(OPTION_TABLE, "person") //Table to read.
>>>>>>> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
>>>>>>> .load()
>>>>>>> .filter(col("id") >= 2) //Filter clause.
>>>>>>> .filter(col("name") like "%J%") //Another filter clause.
>>>>>>>
>>>>>>> println("Data frame schema:")
>>>>>>>
>>>>>>> igniteDF.printSchema() //Printing query schema to console.
>>>>>>>
>>>>>>> println("Data frame content:")
>>>>>>>
>>>>>>> igniteDF.show() //Printing query results to console.
>>>>>>> }
>>>>>>>
>>>>>>>
>>>>>>> def main(args: Array[String]): Unit = {
>>>>>>>
>>>>>>> setupExampleData
>>>>>>>
>>>>>>> //Creating spark session.
>>>>>>> implicit val spark = SparkSession.builder()
>>>>>>> .appName("Spark Ignite data sources example")
>>>>>>> .master("local")
>>>>>>> .config("spark.executor.instances", "2")
>>>>>>> .getOrCreate()
>>>>>>>
>>>>>>> // Adjust the logger to exclude the logs of no interest.
>>>>>>> Logger.getRootLogger.setLevel(Level.ERROR)
>>>>>>> Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
>>>>>>>
>>>>>>> //sparkDSLExample
>>>>>>>
>>>>>>>
>>>>>>> val df = spark.read
>>>>>>> .format("jdbc")
>>>>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>>>>> .option("fetchsize",100)
>>>>>>> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
>>>>>>> .option("dbtable", "Person").load()
>>>>>>>
>>>>>>> df.printSchema()
>>>>>>>
>>>>>>> df.createOrReplaceTempView("test")
>>>>>>>
>>>>>>> spark.sql("select * from test where id=1").show(10)
>>>>>>>
>>>>>>> spark.sql("select 4,'blah',124232").show(10)
>>>>>>>
>>>>>>> import java.sql.DriverManager
>>>>>>> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>>>>>>>
>>>>>>> import java.util.Properties
>>>>>>> val connectionProperties = new Properties()
>>>>>>>
>>>>>>> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>>>>>>>
>>>>>>> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
>>>>>>> "Person",connectionProperties)
>>>>>>>
>>>>>>> spark.read
>>>>>>> .format("jdbc")
>>>>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>>>>> .option("fetchsize",100)
>>>>>>> .option("dbtable", "Person").load().show(10,false)
>>>>>>>
>>>>>>> }
>>>>>>>
>>>>>>> }
>>>>>>>
>>>>>>>
>>>>>>> example-default.xml:-
>>>>>>>
>>>>>>> <?xml version="1.0" encoding="UTF-8"?>
>>>>>>>
>>>>>>> <!--
>>>>>>> Licensed to the Apache Software Foundation (ASF) under one or more
>>>>>>> contributor license agreements. See the NOTICE file distributed with
>>>>>>> this work for additional information regarding copyright ownership.
>>>>>>> The ASF licenses this file to You under the Apache License, Version 2.0
>>>>>>> (the "License"); you may not use this file except in compliance with
>>>>>>> the License. You may obtain a copy of the License at
>>>>>>>
>>>>>>> http://www.apache.org/licenses/LICENSE-2.0
>>>>>>>
>>>>>>> Unless required by applicable law or agreed to in writing, software
>>>>>>> distributed under the License is distributed on an "AS IS" BASIS,
>>>>>>> WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
>>>>>>> See the License for the specific language governing permissions and
>>>>>>> limitations under the License.
>>>>>>> -->
>>>>>>>
>>>>>>> <!--
>>>>>>> Ignite configuration with all defaults and enabled p2p deployment and enabled events.
>>>>>>> -->
>>>>>>> <beans xmlns="http://www.springframework.org/schema/beans"
>>>>>>> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
>>>>>>> xmlns:util="http://www.springframework.org/schema/util"
>>>>>>> xsi:schemaLocation="
>>>>>>> http://www.springframework.org/schema/beans
>>>>>>> http://www.springframework.org/schema/beans/spring-beans.xsd
>>>>>>> http://www.springframework.org/schema/util
>>>>>>> http://www.springframework.org/schema/util/spring-util.xsd">
>>>>>>> <bean abstract="true" id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
>>>>>>> <!-- Set to true to enable distributed class loading for examples, default is false. -->
>>>>>>> <property name="peerClassLoadingEnabled" value="true"/>
>>>>>>>
>>>>>>> <!-- Enable task execution events for examples. -->
>>>>>>> <property name="includeEventTypes">
>>>>>>> <list>
>>>>>>> <!--Task execution events-->
>>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
>>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
>>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
>>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
>>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
>>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
>>>>>>>
>>>>>>> <!--Cache events-->
>>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
>>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
>>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
>>>>>>> </list>
>>>>>>> </property>
>>>>>>>
>>>>>>> <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
>>>>>>> <property name="discoverySpi">
>>>>>>> <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
>>>>>>> <property name="ipFinder">
>>>>>>> <!--
>>>>>>> Ignite provides several options for automatic discovery that can be used
>>>>>>> instead os static IP based discovery. For information on all options refer
>>>>>>> to our documentation: http://apacheignite.readme.io/docs/cluster-config
>>>>>>> -->
>>>>>>> <!-- Uncomment static IP finder to enable static-based discovery of initial nodes. -->
>>>>>>> <!--<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
>>>>>>> <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
>>>>>>> <property name="addresses">
>>>>>>> <list>
>>>>>>> <!-- In distributed environment, replace with actual host IP address. -->
>>>>>>> <value>3.88.248.113:10800</value>
>>>>>>> </list>
>>>>>>> </property>
>>>>>>> </bean>
>>>>>>> </property>
>>>>>>> </bean>
>>>>>>> </property>
>>>>>>> </bean>
>>>>>>> </beans>
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>> --
>>>>>>> Thanks & Regards
>>>>>>> Sri Tummala
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>
>>>>>> --
>>>>>> Thanks & Regards
>>>>>> Sri Tummala
>>>>>>
>>>>>>
>>>>>
>>>>> --
>>>>> Thanks & Regards
>>>>> Sri Tummala
>>>>>
>>>>>
>>>>
>>>> --
>>>> Thanks & Regards
>>>> Sri Tummala
>>>>
>>>>
>>>>
>>>>
>>>
>>> --
>>> Thanks & Regards
>>> Sri Tummala
>>>
>>>
>>
>> --
>> Thanks & Regards
>> Sri Tummala
>>
>>
>
> --
> Thanks & Regards
> Sri Tummala
>
>
Re: Apache Spark + Ignite Connection Issue
Posted by sri hari kali charan Tummala <ka...@gmail.com>.
so I got an answer from Grid Grain computing, JDBC thin client is the only
way yo connect the ignite cluster which is running on aws.
https://forums.gridgain.com/community-home/digestviewer/viewthread?MessageKey=13f5e836-1569-486a-8475-84c70fc141e0&CommunityKey=3b551477-7e2d-462f-bc5f-d6d10ccbbe35&tab=digestviewer&SuccessMsg=Thank+you+for+submitting+your+message.
Thanks
Sri
On Fri, Oct 18, 2019 at 1:38 PM sri hari kali charan Tummala <
kali.tummala@gmail.com> wrote:
> Ok I created a 2 node ec2 instance ignite cluster is below is the right
> way to create cache? the code still using my laptop resources unable to
> connect ignite cluster.
>
> Out put:-
> [13:36:36] Ignite node started OK (id=8535f4d3)
> [13:36:36] Topology snapshot [ver=30, locNode=8535f4d3, *servers=1,*
> clients=1, state=ACTIVE, CPUs=8, *offheap=6.4GB, heap=14.0GB*]
> >>> cache acquired
>
> package com.ignite.examples.igniteStartup
>
> import java.util.Arrays
> import java.util.List
> import com.ignite.examples.model.Address
> import org.apache.commons.logging.Log
> import org.apache.commons.logging.LogFactory
> import org.apache.ignite.Ignite
> import org.apache.ignite.IgniteCache
> import org.apache.ignite.Ignition
> import org.apache.ignite.configuration.IgniteConfiguration
> import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi
> import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder
>
> //remove if not needed
> import scala.collection.JavaConversions._
>
> object IgniteStart2 {
>
> private var cache: IgniteCache[String, Address] = _
>
> def main(args: Array[String]): Unit = {
>
> val spi: TcpDiscoverySpi = new TcpDiscoverySpi()
> val ipFinder: TcpDiscoveryVmIpFinder = new TcpDiscoveryVmIpFinder()
> val hostList: List[String] = Arrays.asList("127.0.0.1:47500..47509,18.206.247.40:47500..47509,52.207.217.31:47500..47509".split(","): _*)
> ipFinder.setAddresses(hostList)
> spi.setIpFinder(ipFinder)
> val cfg: IgniteConfiguration = new IgniteConfiguration()
> cfg.setDiscoverySpi(spi)
> cfg.setClientMode(true)
> System.out.println(">>> I am here")
>
> cfg.setPeerClassLoadingEnabled(true)
> val ignite: Ignite = Ignition.start(cfg)
> cache = Ignition.ignite().cache("test")
> System.out.println(">>> cache acquired")
>
> System.exit(0)
>
>
> }
>
> }
>
>
> On Fri, Oct 18, 2019 at 12:56 PM sri hari kali charan Tummala <
> kali.tummala@gmail.com> wrote:
>
>> Hi Stephen ,
>>
>> I followed below steps and created one node ec2 instance with Ignite on
>> it now I am connecting to the ignite cluster using spark in two ways 1)
>> thin client 2) I don't know you have to tell me
>>
>> my question is what is the value I have to pass for option 2 is it ec2
>> instance public IP with port 47500..47509 in my example-default.xml file ?
>>
>>
>> https://www.gridgain.com/docs/8.7.6//installation-guide/manual-install-on-ec2
>>
>> Thanks
>> Sri
>>
>> On Fri, Oct 18, 2019 at 12:47 PM Stephen Darlington <
>> stephen.darlington@gridgain.com> wrote:
>>
>>> You’re still pointing your Spark node (*thick* client) at port 10800
>>> (the *thin* client port). This is not going to work.
>>>
>>> You can create a table using sqlline and read it using Spark, or vice
>>> versa. But you need a functioning cluster.
>>>
>>> Check out the documentation on clustering concepts and configuration:
>>> https://www.gridgain.com/docs/latest/developers-guide/clustering/clustering
>>>
>>>
>>> On 18 Oct 2019, at 16:32, sri hari kali charan Tummala <
>>> kali.tummala@gmail.com> wrote:
>>>
>>> Hi Stephen/All,
>>>
>>> got it working somewhat using below, but have an issue table and data
>>> which is created using thin client is failing to read using spark but table
>>> created by spark can be read using a thin client, that means table created
>>> in Ignite using spark are the only ones read using spark in Ignite?
>>>
>>> example-default.xml
>>>
>>> <property name="addresses">
>>> <list>
>>> <!-- In distributed environment, replace with actual host IP address. -->
>>> <value>18.206.247.40:10800</value>
>>> </list>
>>> </property>
>>>
>>>
>>> NotWorking Scala Function:-
>>>
>>> def readThinClientTableUsingSpark(implicit spark: SparkSession) = {
>>> val personDataFrame = spark.read
>>> .format(FORMAT_IGNITE)
>>> .option(OPTION_CONFIG_FILE, CONFIG)
>>> .option(OPTION_TABLE, "person")
>>> .load()
>>>
>>> println()
>>> println("Data frame thin connection content:person")
>>> println()
>>>
>>> //Printing content of data frame to console.
>>> personDataFrame.show()
>>> }
>>>
>>>
>>>
>>>
>>>
>>>
>>>
>>>
>>>
>>>
>>>
>>>
>>>
>>> *Exception in thread "main" class org.apache.ignite.IgniteException:
>>> Unknown table person at
>>> org.apache.ignite.spark.impl.IgniteSQLRelation$$anonfun$schema$2.apply(IgniteSQLRelation.scala:46)
>>> at
>>> org.apache.ignite.spark.impl.IgniteSQLRelation$$anonfun$schema$2.apply(IgniteSQLRelation.scala:46)
>>> at scala.Option.getOrElse(Option.scala:121) at
>>> org.apache.ignite.spark.impl.IgniteSQLRelation.schema(IgniteSQLRelation.scala:46)
>>> at
>>> org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:431)
>>> at
>>> org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:239)
>>> at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:227) at
>>> org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:164) at
>>> com.ignite.examples.spark.SparkIgniteCleanCode$.readThinClientTableUsingSpark(SparkIgniteCleanCode.scala:154)
>>> at
>>> com.ignite.examples.spark.SparkIgniteCleanCode$.main(SparkIgniteCleanCode.scala:216)
>>> at
>>> com.ignite.examples.spark.SparkIgniteCleanCode.main(SparkIgniteCleanCode.scala)*
>>>
>>> *Full Code:- (step 7 Fails)*
>>>
>>> package com.ignite.examples.spark
>>>
>>> import com.ignite.examples.model.Address
>>> import org.apache.ignite.{Ignite, Ignition}
>>> import org.apache.ignite.cache.query.SqlFieldsQuery
>>> import org.apache.ignite.client.{ClientCache, IgniteClient}
>>> import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
>>> import java.lang.{Long => JLong, String => JString}
>>>
>>> import com.ignite.examples.spark.SparkClientConnectionTest.{CACHE_NAME, CONFIG}
>>> import org.apache.ignite.internal.util.IgniteUtils.resolveIgnitePath
>>> import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE, OPTION_CONFIG_FILE, OPTION_CREATE_TABLE_PARAMETERS, OPTION_CREATE_TABLE_PRIMARY_KEY_FIELDS, OPTION_TABLE}
>>> import org.apache.log4j.{Level, Logger}
>>> import org.apache.spark.sql.{SaveMode, SparkSession}
>>>
>>> object SparkIgniteCleanCode {
>>>
>>> private val CACHE_NAME = "SparkCache"
>>>
>>> private val CONFIG = "/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
>>>
>>> def setupExampleData = {
>>>
>>> val cfg2 = new ClientConfiguration().setAddresses("18.206.247.40:10800")
>>> val igniteClient:IgniteClient = Ignition.startClient(cfg2)
>>>
>>> System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
>>>
>>> val cache:ClientCache[Integer, Address] = igniteClient.getOrCreateCache(CACHE_NAME)
>>>
>>> cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
>>> .setSchema("PUBLIC")).getAll
>>>
>>> cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) ) WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
>>> .setSchema("PUBLIC")).getAll
>>>
>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco", "04074").setSchema("PUBLIC")).getAll
>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road", "520003").setSchema("PUBLIC")).getAll
>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road", "1234").setSchema("PUBLIC")).getAll
>>>
>>> System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
>>>
>>> val data=cache.query(new SqlFieldsQuery("select * from Person").setSchema("PUBLIC")).getAll
>>>
>>> println(data.toString)
>>> }
>>>
>>> def sparkReadIgniteWithThinClient(implicit spark: SparkSession)={
>>>
>>> val df = spark.read
>>> .format("jdbc")
>>> .option("url", "jdbc:ignite:thin://18.206.247.40")
>>> .option("fetchsize",100)
>>> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
>>> .option("dbtable", "Person").load()
>>>
>>> df.printSchema()
>>>
>>> df.createOrReplaceTempView("test")
>>>
>>> spark.sql("select * from test where id=1").show(10)
>>>
>>> spark.sql("select 4,'blah',124232").show(10)
>>>
>>> }
>>>
>>> def sparkWriteIgniteWithThinClient(implicit spark: SparkSession)={
>>>
>>> import java.sql.DriverManager
>>> val connection = DriverManager.getConnection("jdbc:ignite:thin://18.206.247.40")
>>>
>>> import java.util.Properties
>>> val connectionProperties = new Properties()
>>>
>>> connectionProperties.put("url", "jdbc:ignite:thin://18.206.247.40")
>>>
>>> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://18.206.247.40",
>>> "Person",connectionProperties)
>>>
>>> spark.read
>>> .format("jdbc")
>>> .option("url", "jdbc:ignite:thin://18.206.247.40")
>>> .option("fetchsize",100)
>>> .option("dbtable", "Person").load().show(10,false)
>>>
>>> }
>>>
>>> def writeJSonToIgniteUsingSpark(implicit spark: SparkSession): Unit = {
>>>
>>> val ignite = Ignition.start(CONFIG)
>>>
>>> //Load content of json file to data frame.
>>> val personsDataFrame = spark.read.json(
>>> resolveIgnitePath("/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/resources/person.json").getAbsolutePath)
>>>
>>> println()
>>> println("Json file content:")
>>> println()
>>>
>>> //Printing content of json file to console.
>>> personsDataFrame.show()
>>>
>>> println()
>>> println("Writing Data Frame to Ignite:")
>>> println()
>>>
>>> //Writing content of data frame to Ignite.
>>> personsDataFrame.write
>>> .format(FORMAT_IGNITE)
>>> .mode(org.apache.spark.sql.SaveMode.Append)
>>> .option(OPTION_CONFIG_FILE, CONFIG)
>>> .option(OPTION_TABLE, "json_person")
>>> .option(OPTION_CREATE_TABLE_PRIMARY_KEY_FIELDS, "id")
>>> .option(OPTION_CREATE_TABLE_PARAMETERS, "template=replicated")
>>> .save()
>>>
>>> println("Done!")
>>>
>>> println()
>>> println("Reading data from Ignite table:")
>>> println()
>>>
>>> val cache = ignite.cache[Any, Any](CACHE_NAME)
>>>
>>> //Reading saved data from Ignite.
>>> val data = cache.query(new SqlFieldsQuery("SELECT id, name, department FROM json_person")).getAll
>>>
>>> println(data.toString)
>>>
>>> //data.foreach { row ⇒ println(row.mkString("[", ", ", "]")) }
>>> }
>>>
>>> def readIgniteUsingSpark(implicit spark: SparkSession) = {
>>> val json_person = spark.read
>>> .format(FORMAT_IGNITE)
>>> .option(OPTION_CONFIG_FILE, CONFIG)
>>> .option(OPTION_TABLE, "json_person")
>>> .load()
>>>
>>> println()
>>> println("Data frame content:json_person")
>>> println()
>>>
>>> //Printing content of data frame to console.
>>> json_person.show()
>>> }
>>>
>>>
>>> def readThinClientTableUsingSpark(implicit spark: SparkSession) = {
>>> val personDataFrame = spark.read
>>> .format(FORMAT_IGNITE)
>>> .option(OPTION_CONFIG_FILE, CONFIG)
>>> .option(OPTION_TABLE, "person")
>>> .load()
>>>
>>> println()
>>> println("Data frame thin connection content:person")
>>> println()
>>>
>>> //Printing content of data frame to console.
>>> personDataFrame.show()
>>> }
>>>
>>>
>>> def main(args: Array[String]): Unit = {
>>>
>>> println()
>>> println("Step 1 setupExampleData:")
>>> println()
>>>
>>> setupExampleData
>>>
>>> println()
>>> println("Step 2 createSparkSession:")
>>> println()
>>>
>>> //Creating spark session.
>>> implicit val spark = SparkSession.builder()
>>> .appName("Spark Ignite data sources example")
>>> .master("local")
>>> .config("spark.executor.instances", "2")
>>> .getOrCreate()
>>>
>>> // Adjust the logger to exclude the logs of no interest.
>>> Logger.getRootLogger.setLevel(Level.ERROR)
>>> Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
>>>
>>> println()
>>> println("Step 3 ReadIgniteWithThinClient of Step1 Data:")
>>> println()
>>>
>>> sparkReadIgniteWithThinClient(spark)
>>>
>>> println()
>>> println("Step 4 sparkWriteIgniteWithThinClient of Step1 Data:")
>>> println()
>>>
>>> sparkWriteIgniteWithThinClient(spark)
>>>
>>> println()
>>> println("Step 5 writeJSonToIgniteUsingSpark Using Spark:")
>>> println()
>>>
>>> writeJSonToIgniteUsingSpark(spark)
>>>
>>> println()
>>> println("Step 6 readIgniteUsingSpark Using Spark:")
>>> println()
>>>
>>> readIgniteUsingSpark(spark)
>>>
>>> println()
>>> println("Step 7 readThinClientTableUsingSpark Using Spark:")
>>> println()
>>>
>>> readThinClientTableUsingSpark(spark)
>>>
>>>
>>> spark.stop()
>>>
>>>
>>> }
>>>
>>> }
>>>
>>>
>>> example-default.xml
>>>
>>> <?xml version="1.0" encoding="UTF-8"?>
>>>
>>> <!--
>>> Licensed to the Apache Software Foundation (ASF) under one or more
>>> contributor license agreements. See the NOTICE file distributed with
>>> this work for additional information regarding copyright ownership.
>>> The ASF licenses this file to You under the Apache License, Version 2.0
>>> (the "License"); you may not use this file except in compliance with
>>> the License. You may obtain a copy of the License at
>>>
>>> http://www.apache.org/licenses/LICENSE-2.0
>>>
>>> Unless required by applicable law or agreed to in writing, software
>>> distributed under the License is distributed on an "AS IS" BASIS,
>>> WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
>>> See the License for the specific language governing permissions and
>>> limitations under the License.
>>> -->
>>>
>>> <!--
>>> Ignite configuration with all defaults and enabled p2p deployment and enabled events.
>>> -->
>>> <beans xmlns="http://www.springframework.org/schema/beans"
>>> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
>>> xmlns:util="http://www.springframework.org/schema/util"
>>> xsi:schemaLocation="
>>> http://www.springframework.org/schema/beans
>>> http://www.springframework.org/schema/beans/spring-beans.xsd
>>> http://www.springframework.org/schema/util
>>> http://www.springframework.org/schema/util/spring-util.xsd">
>>> <bean abstract="true" id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
>>> <!-- Set to true to enable distributed class loading for examples, default is false. -->
>>> <property name="peerClassLoadingEnabled" value="true"/>
>>>
>>> <!-- Enable task execution events for examples. -->
>>> <property name="includeEventTypes">
>>> <list>
>>> <!--Task execution events-->
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
>>>
>>> <!--Cache events-->
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
>>> </list>
>>> </property>
>>>
>>> <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
>>> <property name="discoverySpi">
>>> <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
>>> <property name="ipFinder">
>>> <!--
>>> Ignite provides several options for automatic discovery that can be used
>>> instead os static IP based discovery. For information on all options refer
>>> to our documentation: http://apacheignite.readme.io/docs/cluster-config
>>> -->
>>> <!-- Uncomment static IP finder to enable static-based discovery of initial nodes. -->
>>> <!--<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
>>> <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
>>> <property name="addresses">
>>> <list>
>>> <!-- In distributed environment, replace with actual host IP address. -->
>>> <value>18.206.247.40:10800</value>
>>> </list>
>>> </property>
>>> </bean>
>>> </property>
>>> </bean>
>>> </property>
>>> </bean>
>>> </beans>
>>>
>>>
>>> Thanks
>>> Sri
>>>
>>> On Fri, Oct 18, 2019 at 9:23 AM sri hari kali charan Tummala <
>>> kali.tummala@gmail.com> wrote:
>>>
>>>> do you mean communication ports 47100-47200 as mentioned (
>>>> https://www.gridgain.com/docs/8.7.6//installation-guide/manual-install-on-ec2)
>>>> ? bare in mind I am running my spark job outside ignite ec2 box (MAC PC).
>>>>
>>>> which option is right in my default.xml file?
>>>>
>>>> Option 1:-
>>>>
>>>> <property name="addresses">
>>>> <list>
>>>> <!-- In distributed environment, replace with actual host IP address. -->
>>>> <value>3.88.248.113:4 <http://3.88.248.113:10800/>7100..47200</value>
>>>> </list>
>>>> </property>
>>>>
>>>>
>>>> Option 2:-
>>>>
>>>> <property name="addresses">
>>>> <list>
>>>> <!-- In distributed environment, replace with actual host IP address. -->
>>>> <value>3.88.248.113:4 <http://3.88.248.113:10800/>7500..47600</value>
>>>> </list>
>>>> </property>
>>>>
>>>>
>>>> Option 3:-
>>>>
>>>> <property name="addresses">
>>>> <list>
>>>> <!-- In distributed environment, replace with actual host IP address. -->
>>>> <value>3.88.248.113 <http://3.88.248.113:10800/></value>
>>>> </list>
>>>> </property>
>>>>
>>>>
>>>>
>>>> Thanks
>>>>
>>>> On Fri, Oct 18, 2019 at 9:18 AM sri hari kali charan Tummala <
>>>> kali.tummala@gmail.com> wrote:
>>>>
>>>>> Hi Stephen ,
>>>>>
>>>>> do you mean 3.88.248.113: <http://3.88.248.113:10800/>47500..47700
>>>>> something like this? or just public ip 3.88.248.113
>>>>> <http://3.88.248.113:10800/> I tried all the possibilities none of
>>>>> them are getting connected.
>>>>>
>>>>> Thanks
>>>>> Sri
>>>>>
>>>>> On Fri, Oct 18, 2019 at 6:02 AM Stephen Darlington <
>>>>> stephen.darlington@gridgain.com> wrote:
>>>>>
>>>>>> You’re trying to connect a thick client (the Spark integration) to
>>>>>> the thin client port (10800). Your example-default.xml file needs to have
>>>>>> the same configuration as your server node(s).
>>>>>>
>>>>>> Regards,
>>>>>> Stephen
>>>>>>
>>>>>> On 17 Oct 2019, at 18:12, sri hari kali charan Tummala <
>>>>>> kali.tummala@gmail.com> wrote:
>>>>>>
>>>>>> Hi Community,
>>>>>>
>>>>>> I am trying to read and write into the Ignite cluster using
>>>>>> apache-spark I am able to do that using JDBC thin client but not native
>>>>>> method as mentioned in several spark + ignite examples.
>>>>>>
>>>>>> Right now all the spark + ignite examples launch a local ignite
>>>>>> cluster but I want my code connecting to already existing cluster (client).
>>>>>>
>>>>>> Question:-
>>>>>> *How to pass Ignite connection ip and port (10800) 10800 in
>>>>>> example-default.xml ?*
>>>>>>
>>>>>> Error:-
>>>>>> *TcpDiscoverySpi: Failed to connect to any address from IP finder
>>>>>> (will retry to join topology every 2000 ms; change 'reconnectDelay' to
>>>>>> configure the frequency of retries): [/3.88.248.113:10800
>>>>>> <http://3.88.248.113:10800/>]*
>>>>>>
>>>>>> Working (Spark + Ignite using JDBC):-
>>>>>>
>>>>>> val df = spark.read
>>>>>> .format("jdbc")
>>>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>>>> .option("fetchsize",100)
>>>>>> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
>>>>>> .option("dbtable", "Person").load()
>>>>>>
>>>>>> df.printSchema()
>>>>>>
>>>>>> df.createOrReplaceTempView("test")
>>>>>>
>>>>>> spark.sql("select * from test where id=1").show(10)
>>>>>>
>>>>>> spark.sql("select 4,'blah',124232").show(10)
>>>>>>
>>>>>> import java.sql.DriverManager
>>>>>> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>>>>>>
>>>>>> import java.util.Properties
>>>>>> val connectionProperties = new Properties()
>>>>>>
>>>>>> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>>>>>>
>>>>>> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
>>>>>> "Person",connectionProperties)
>>>>>>
>>>>>> spark.read
>>>>>> .format("jdbc")
>>>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>>>> .option("fetchsize",100)
>>>>>> .option("dbtable", "Person").load().show(10,false)
>>>>>>
>>>>>>
>>>>>> Not Working requires a CONFIG file which is example-default.xml:-
>>>>>>
>>>>>> val igniteDF = spark.read
>>>>>> .format(FORMAT_IGNITE) //Data source type.
>>>>>> .option(OPTION_TABLE, "person") //Table to read.
>>>>>> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
>>>>>> .load()
>>>>>> .filter(col("id") >= 2) //Filter clause.
>>>>>> .filter(col("name") like "%J%") //Another filter clause.
>>>>>>
>>>>>>
>>>>>> Full Code:- (sparkDSLExample) function fails to connect ignite
>>>>>> cluster which I already have
>>>>>>
>>>>>> package com.ignite.examples.spark
>>>>>>
>>>>>> import com.ignite.examples.model.Address
>>>>>> import org.apache.ignite.{Ignite, Ignition}
>>>>>> import org.apache.ignite.cache.query.SqlFieldsQuery
>>>>>> import org.apache.ignite.client.{ClientCache, IgniteClient}
>>>>>> import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
>>>>>> import java.lang.{Long => JLong, String => JString}
>>>>>>
>>>>>> import org.apache.ignite.cache.query.SqlFieldsQuery
>>>>>> import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE, OPTION_CONFIG_FILE, OPTION_TABLE}
>>>>>> import org.apache.log4j.{Level, Logger}
>>>>>> import org.apache.spark.sql.{SaveMode, SparkSession}
>>>>>> import org.apache.spark.sql.functions.col
>>>>>>
>>>>>> object SparkClientConnectionTest {
>>>>>>
>>>>>> private val CACHE_NAME = "SparkCache"
>>>>>>
>>>>>> private val CONFIG = "/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
>>>>>>
>>>>>> def setupExampleData = {
>>>>>>
>>>>>> val cfg2 = new ClientConfiguration().setAddresses("3.88.248.113:10800")
>>>>>> val igniteClient:IgniteClient = Ignition.startClient(cfg2)
>>>>>>
>>>>>> System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
>>>>>>
>>>>>> val cache:ClientCache[Integer, Address] = igniteClient.getOrCreateCache(CACHE_NAME)
>>>>>>
>>>>>> cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
>>>>>> .setSchema("PUBLIC")).getAll
>>>>>>
>>>>>> cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) ) WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
>>>>>> .setSchema("PUBLIC")).getAll
>>>>>>
>>>>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco", "04074").setSchema("PUBLIC")).getAll
>>>>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road", "520003").setSchema("PUBLIC")).getAll
>>>>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road", "1234").setSchema("PUBLIC")).getAll
>>>>>>
>>>>>> System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
>>>>>>
>>>>>> val data=cache.query(new SqlFieldsQuery("select * from Person").setSchema("PUBLIC")).getAll
>>>>>>
>>>>>> println(data.toString)
>>>>>> }
>>>>>>
>>>>>> def sparkDSLExample(implicit spark: SparkSession): Unit = {
>>>>>> println("Querying using Spark DSL.")
>>>>>> println
>>>>>>
>>>>>>
>>>>>> val igniteDF = spark.read
>>>>>> .format(FORMAT_IGNITE) //Data source type.
>>>>>> .option(OPTION_TABLE, "person") //Table to read.
>>>>>> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
>>>>>> .load()
>>>>>> .filter(col("id") >= 2) //Filter clause.
>>>>>> .filter(col("name") like "%J%") //Another filter clause.
>>>>>>
>>>>>> println("Data frame schema:")
>>>>>>
>>>>>> igniteDF.printSchema() //Printing query schema to console.
>>>>>>
>>>>>> println("Data frame content:")
>>>>>>
>>>>>> igniteDF.show() //Printing query results to console.
>>>>>> }
>>>>>>
>>>>>>
>>>>>> def main(args: Array[String]): Unit = {
>>>>>>
>>>>>> setupExampleData
>>>>>>
>>>>>> //Creating spark session.
>>>>>> implicit val spark = SparkSession.builder()
>>>>>> .appName("Spark Ignite data sources example")
>>>>>> .master("local")
>>>>>> .config("spark.executor.instances", "2")
>>>>>> .getOrCreate()
>>>>>>
>>>>>> // Adjust the logger to exclude the logs of no interest.
>>>>>> Logger.getRootLogger.setLevel(Level.ERROR)
>>>>>> Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
>>>>>>
>>>>>> //sparkDSLExample
>>>>>>
>>>>>>
>>>>>> val df = spark.read
>>>>>> .format("jdbc")
>>>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>>>> .option("fetchsize",100)
>>>>>> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
>>>>>> .option("dbtable", "Person").load()
>>>>>>
>>>>>> df.printSchema()
>>>>>>
>>>>>> df.createOrReplaceTempView("test")
>>>>>>
>>>>>> spark.sql("select * from test where id=1").show(10)
>>>>>>
>>>>>> spark.sql("select 4,'blah',124232").show(10)
>>>>>>
>>>>>> import java.sql.DriverManager
>>>>>> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>>>>>>
>>>>>> import java.util.Properties
>>>>>> val connectionProperties = new Properties()
>>>>>>
>>>>>> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>>>>>>
>>>>>> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
>>>>>> "Person",connectionProperties)
>>>>>>
>>>>>> spark.read
>>>>>> .format("jdbc")
>>>>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>>>>> .option("fetchsize",100)
>>>>>> .option("dbtable", "Person").load().show(10,false)
>>>>>>
>>>>>> }
>>>>>>
>>>>>> }
>>>>>>
>>>>>>
>>>>>> example-default.xml:-
>>>>>>
>>>>>> <?xml version="1.0" encoding="UTF-8"?>
>>>>>>
>>>>>> <!--
>>>>>> Licensed to the Apache Software Foundation (ASF) under one or more
>>>>>> contributor license agreements. See the NOTICE file distributed with
>>>>>> this work for additional information regarding copyright ownership.
>>>>>> The ASF licenses this file to You under the Apache License, Version 2.0
>>>>>> (the "License"); you may not use this file except in compliance with
>>>>>> the License. You may obtain a copy of the License at
>>>>>>
>>>>>> http://www.apache.org/licenses/LICENSE-2.0
>>>>>>
>>>>>> Unless required by applicable law or agreed to in writing, software
>>>>>> distributed under the License is distributed on an "AS IS" BASIS,
>>>>>> WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
>>>>>> See the License for the specific language governing permissions and
>>>>>> limitations under the License.
>>>>>> -->
>>>>>>
>>>>>> <!--
>>>>>> Ignite configuration with all defaults and enabled p2p deployment and enabled events.
>>>>>> -->
>>>>>> <beans xmlns="http://www.springframework.org/schema/beans"
>>>>>> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
>>>>>> xmlns:util="http://www.springframework.org/schema/util"
>>>>>> xsi:schemaLocation="
>>>>>> http://www.springframework.org/schema/beans
>>>>>> http://www.springframework.org/schema/beans/spring-beans.xsd
>>>>>> http://www.springframework.org/schema/util
>>>>>> http://www.springframework.org/schema/util/spring-util.xsd">
>>>>>> <bean abstract="true" id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
>>>>>> <!-- Set to true to enable distributed class loading for examples, default is false. -->
>>>>>> <property name="peerClassLoadingEnabled" value="true"/>
>>>>>>
>>>>>> <!-- Enable task execution events for examples. -->
>>>>>> <property name="includeEventTypes">
>>>>>> <list>
>>>>>> <!--Task execution events-->
>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
>>>>>>
>>>>>> <!--Cache events-->
>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
>>>>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
>>>>>> </list>
>>>>>> </property>
>>>>>>
>>>>>> <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
>>>>>> <property name="discoverySpi">
>>>>>> <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
>>>>>> <property name="ipFinder">
>>>>>> <!--
>>>>>> Ignite provides several options for automatic discovery that can be used
>>>>>> instead os static IP based discovery. For information on all options refer
>>>>>> to our documentation: http://apacheignite.readme.io/docs/cluster-config
>>>>>> -->
>>>>>> <!-- Uncomment static IP finder to enable static-based discovery of initial nodes. -->
>>>>>> <!--<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
>>>>>> <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
>>>>>> <property name="addresses">
>>>>>> <list>
>>>>>> <!-- In distributed environment, replace with actual host IP address. -->
>>>>>> <value>3.88.248.113:10800</value>
>>>>>> </list>
>>>>>> </property>
>>>>>> </bean>
>>>>>> </property>
>>>>>> </bean>
>>>>>> </property>
>>>>>> </bean>
>>>>>> </beans>
>>>>>>
>>>>>>
>>>>>>
>>>>>>
>>>>>> --
>>>>>> Thanks & Regards
>>>>>> Sri Tummala
>>>>>>
>>>>>>
>>>>>>
>>>>>>
>>>>>
>>>>> --
>>>>> Thanks & Regards
>>>>> Sri Tummala
>>>>>
>>>>>
>>>>
>>>> --
>>>> Thanks & Regards
>>>> Sri Tummala
>>>>
>>>>
>>>
>>> --
>>> Thanks & Regards
>>> Sri Tummala
>>>
>>>
>>>
>>>
>>
>> --
>> Thanks & Regards
>> Sri Tummala
>>
>>
>
> --
> Thanks & Regards
> Sri Tummala
>
>
--
Thanks & Regards
Sri Tummala
Re: Apache Spark + Ignite Connection Issue
Posted by sri hari kali charan Tummala <ka...@gmail.com>.
Hi Stephen/All,
got it working somewhat using below, but have an issue table and data which
is created using thin client is failing to read using spark but table
created by spark can be read using a thin client, that means table created
in Ignite using spark are the only ones read using spark in Ignite?
example-default.xml
<property name="addresses">
<list>
<!-- In distributed environment, replace with actual host IP
address. -->
<value>18.206.247.40:10800</value>
</list>
</property>
NotWorking Scala Function:-
def readThinClientTableUsingSpark(implicit spark: SparkSession) = {
val personDataFrame = spark.read
.format(FORMAT_IGNITE)
.option(OPTION_CONFIG_FILE, CONFIG)
.option(OPTION_TABLE, "person")
.load()
println()
println("Data frame thin connection content:person")
println()
//Printing content of data frame to console.
personDataFrame.show()
}
*Exception in thread "main" class org.apache.ignite.IgniteException:
Unknown table person at
org.apache.ignite.spark.impl.IgniteSQLRelation$$anonfun$schema$2.apply(IgniteSQLRelation.scala:46)
at
org.apache.ignite.spark.impl.IgniteSQLRelation$$anonfun$schema$2.apply(IgniteSQLRelation.scala:46)
at scala.Option.getOrElse(Option.scala:121) at
org.apache.ignite.spark.impl.IgniteSQLRelation.schema(IgniteSQLRelation.scala:46)
at
org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:431)
at
org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:239)
at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:227) at
org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:164) at
com.ignite.examples.spark.SparkIgniteCleanCode$.readThinClientTableUsingSpark(SparkIgniteCleanCode.scala:154)
at
com.ignite.examples.spark.SparkIgniteCleanCode$.main(SparkIgniteCleanCode.scala:216)
at
com.ignite.examples.spark.SparkIgniteCleanCode.main(SparkIgniteCleanCode.scala)*
*Full Code:- (step 7 Fails)*
package com.ignite.examples.spark
import com.ignite.examples.model.Address
import org.apache.ignite.{Ignite, Ignition}
import org.apache.ignite.cache.query.SqlFieldsQuery
import org.apache.ignite.client.{ClientCache, IgniteClient}
import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
import java.lang.{Long => JLong, String => JString}
import com.ignite.examples.spark.SparkClientConnectionTest.{CACHE_NAME, CONFIG}
import org.apache.ignite.internal.util.IgniteUtils.resolveIgnitePath
import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE,
OPTION_CONFIG_FILE, OPTION_CREATE_TABLE_PARAMETERS,
OPTION_CREATE_TABLE_PRIMARY_KEY_FIELDS, OPTION_TABLE}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{SaveMode, SparkSession}
object SparkIgniteCleanCode {
private val CACHE_NAME = "SparkCache"
private val CONFIG =
"/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
def setupExampleData = {
val cfg2 = new ClientConfiguration().setAddresses("18.206.247.40:10800")
val igniteClient:IgniteClient = Ignition.startClient(cfg2)
System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
val cache:ClientCache[Integer, Address] =
igniteClient.getOrCreateCache(CACHE_NAME)
cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
.setSchema("PUBLIC")).getAll
cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT
EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) )
WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
.setSchema("PUBLIC")).getAll
cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip)
VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco",
"04074").setSchema("PUBLIC")).getAll
cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip)
VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road",
"520003").setSchema("PUBLIC")).getAll
cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip)
VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road",
"1234").setSchema("PUBLIC")).getAll
System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
val data=cache.query(new SqlFieldsQuery("select * from
Person").setSchema("PUBLIC")).getAll
println(data.toString)
}
def sparkReadIgniteWithThinClient(implicit spark: SparkSession)={
val df = spark.read
.format("jdbc")
.option("url", "jdbc:ignite:thin://18.206.247.40")
.option("fetchsize",100)
//.option("driver", "org.apache.ignite.IgniteJdbcDriver")
.option("dbtable", "Person").load()
df.printSchema()
df.createOrReplaceTempView("test")
spark.sql("select * from test where id=1").show(10)
spark.sql("select 4,'blah',124232").show(10)
}
def sparkWriteIgniteWithThinClient(implicit spark: SparkSession)={
import java.sql.DriverManager
val connection =
DriverManager.getConnection("jdbc:ignite:thin://18.206.247.40")
import java.util.Properties
val connectionProperties = new Properties()
connectionProperties.put("url", "jdbc:ignite:thin://18.206.247.40")
spark.sql("select 4 as ID,'blah' as STREET,124232 as
ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://18.206.247.40",
"Person",connectionProperties)
spark.read
.format("jdbc")
.option("url", "jdbc:ignite:thin://18.206.247.40")
.option("fetchsize",100)
.option("dbtable", "Person").load().show(10,false)
}
def writeJSonToIgniteUsingSpark(implicit spark: SparkSession): Unit = {
val ignite = Ignition.start(CONFIG)
//Load content of json file to data frame.
val personsDataFrame = spark.read.json(
resolveIgnitePath("/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/resources/person.json").getAbsolutePath)
println()
println("Json file content:")
println()
//Printing content of json file to console.
personsDataFrame.show()
println()
println("Writing Data Frame to Ignite:")
println()
//Writing content of data frame to Ignite.
personsDataFrame.write
.format(FORMAT_IGNITE)
.mode(org.apache.spark.sql.SaveMode.Append)
.option(OPTION_CONFIG_FILE, CONFIG)
.option(OPTION_TABLE, "json_person")
.option(OPTION_CREATE_TABLE_PRIMARY_KEY_FIELDS, "id")
.option(OPTION_CREATE_TABLE_PARAMETERS, "template=replicated")
.save()
println("Done!")
println()
println("Reading data from Ignite table:")
println()
val cache = ignite.cache[Any, Any](CACHE_NAME)
//Reading saved data from Ignite.
val data = cache.query(new SqlFieldsQuery("SELECT id, name,
department FROM json_person")).getAll
println(data.toString)
//data.foreach { row ⇒ println(row.mkString("[", ", ", "]")) }
}
def readIgniteUsingSpark(implicit spark: SparkSession) = {
val json_person = spark.read
.format(FORMAT_IGNITE)
.option(OPTION_CONFIG_FILE, CONFIG)
.option(OPTION_TABLE, "json_person")
.load()
println()
println("Data frame content:json_person")
println()
//Printing content of data frame to console.
json_person.show()
}
def readThinClientTableUsingSpark(implicit spark: SparkSession) = {
val personDataFrame = spark.read
.format(FORMAT_IGNITE)
.option(OPTION_CONFIG_FILE, CONFIG)
.option(OPTION_TABLE, "person")
.load()
println()
println("Data frame thin connection content:person")
println()
//Printing content of data frame to console.
personDataFrame.show()
}
def main(args: Array[String]): Unit = {
println()
println("Step 1 setupExampleData:")
println()
setupExampleData
println()
println("Step 2 createSparkSession:")
println()
//Creating spark session.
implicit val spark = SparkSession.builder()
.appName("Spark Ignite data sources example")
.master("local")
.config("spark.executor.instances", "2")
.getOrCreate()
// Adjust the logger to exclude the logs of no interest.
Logger.getRootLogger.setLevel(Level.ERROR)
Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
println()
println("Step 3 ReadIgniteWithThinClient of Step1 Data:")
println()
sparkReadIgniteWithThinClient(spark)
println()
println("Step 4 sparkWriteIgniteWithThinClient of Step1 Data:")
println()
sparkWriteIgniteWithThinClient(spark)
println()
println("Step 5 writeJSonToIgniteUsingSpark Using Spark:")
println()
writeJSonToIgniteUsingSpark(spark)
println()
println("Step 6 readIgniteUsingSpark Using Spark:")
println()
readIgniteUsingSpark(spark)
println()
println("Step 7 readThinClientTableUsingSpark Using Spark:")
println()
readThinClientTableUsingSpark(spark)
spark.stop()
}
}
example-default.xml
<?xml version="1.0" encoding="UTF-8"?>
<!--
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
this work for additional information regarding copyright ownership.
The ASF licenses this file to You under the Apache License, Version 2.0
(the "License"); you may not use this file except in compliance with
the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
-->
<!--
Ignite configuration with all defaults and enabled p2p deployment
and enabled events.
-->
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:util="http://www.springframework.org/schema/util"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/util
http://www.springframework.org/schema/util/spring-util.xsd">
<bean abstract="true" id="ignite.cfg"
class="org.apache.ignite.configuration.IgniteConfiguration">
<!-- Set to true to enable distributed class loading for
examples, default is false. -->
<property name="peerClassLoadingEnabled" value="true"/>
<!-- Enable task execution events for examples. -->
<property name="includeEventTypes">
<list>
<!--Task execution events-->
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
<!--Cache events-->
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
<util:constant
static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
</list>
</property>
<!-- Explicitly configure TCP discovery SPI to provide list of
initial nodes. -->
<property name="discoverySpi">
<bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
<property name="ipFinder">
<!--
Ignite provides several options for automatic
discovery that can be used
instead os static IP based discovery. For
information on all options refer
to our documentation:
http://apacheignite.readme.io/docs/cluster-config
-->
<!-- Uncomment static IP finder to enable
static-based discovery of initial nodes. -->
<!--<bean
class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
<bean
class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
<property name="addresses">
<list>
<!-- In distributed environment,
replace with actual host IP address. -->
<value>18.206.247.40:10800</value>
</list>
</property>
</bean>
</property>
</bean>
</property>
</bean>
</beans>
Thanks
Sri
On Fri, Oct 18, 2019 at 9:23 AM sri hari kali charan Tummala <
kali.tummala@gmail.com> wrote:
> do you mean communication ports 47100-47200 as mentioned (
> https://www.gridgain.com/docs/8.7.6//installation-guide/manual-install-on-ec2)
> ? bare in mind I am running my spark job outside ignite ec2 box (MAC PC).
>
> which option is right in my default.xml file?
>
> Option 1:-
>
> <property name="addresses">
> <list>
> <!-- In distributed environment, replace with actual host IP address. -->
> <value>3.88.248.113:4 <http://3.88.248.113:10800/>7100..47200</value>
> </list>
> </property>
>
>
> Option 2:-
>
> <property name="addresses">
> <list>
> <!-- In distributed environment, replace with actual host IP address. -->
> <value>3.88.248.113:4 <http://3.88.248.113:10800/>7500..47600</value>
> </list>
> </property>
>
>
> Option 3:-
>
> <property name="addresses">
> <list>
> <!-- In distributed environment, replace with actual host IP address. -->
> <value>3.88.248.113 <http://3.88.248.113:10800/></value>
> </list>
> </property>
>
>
>
> Thanks
>
> On Fri, Oct 18, 2019 at 9:18 AM sri hari kali charan Tummala <
> kali.tummala@gmail.com> wrote:
>
>> Hi Stephen ,
>>
>> do you mean 3.88.248.113: <http://3.88.248.113:10800/>47500..47700
>> something like this? or just public ip 3.88.248.113
>> <http://3.88.248.113:10800/> I tried all the possibilities none of them
>> are getting connected.
>>
>> Thanks
>> Sri
>>
>> On Fri, Oct 18, 2019 at 6:02 AM Stephen Darlington <
>> stephen.darlington@gridgain.com> wrote:
>>
>>> You’re trying to connect a thick client (the Spark integration) to the
>>> thin client port (10800). Your example-default.xml file needs to have the
>>> same configuration as your server node(s).
>>>
>>> Regards,
>>> Stephen
>>>
>>> On 17 Oct 2019, at 18:12, sri hari kali charan Tummala <
>>> kali.tummala@gmail.com> wrote:
>>>
>>> Hi Community,
>>>
>>> I am trying to read and write into the Ignite cluster using apache-spark
>>> I am able to do that using JDBC thin client but not native method as
>>> mentioned in several spark + ignite examples.
>>>
>>> Right now all the spark + ignite examples launch a local ignite cluster
>>> but I want my code connecting to already existing cluster (client).
>>>
>>> Question:-
>>> *How to pass Ignite connection ip and port (10800) 10800 in
>>> example-default.xml ?*
>>>
>>> Error:-
>>> *TcpDiscoverySpi: Failed to connect to any address from IP finder (will
>>> retry to join topology every 2000 ms; change 'reconnectDelay' to configure
>>> the frequency of retries): [/3.88.248.113:10800
>>> <http://3.88.248.113:10800/>]*
>>>
>>> Working (Spark + Ignite using JDBC):-
>>>
>>> val df = spark.read
>>> .format("jdbc")
>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>> .option("fetchsize",100)
>>> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
>>> .option("dbtable", "Person").load()
>>>
>>> df.printSchema()
>>>
>>> df.createOrReplaceTempView("test")
>>>
>>> spark.sql("select * from test where id=1").show(10)
>>>
>>> spark.sql("select 4,'blah',124232").show(10)
>>>
>>> import java.sql.DriverManager
>>> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>>>
>>> import java.util.Properties
>>> val connectionProperties = new Properties()
>>>
>>> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>>>
>>> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
>>> "Person",connectionProperties)
>>>
>>> spark.read
>>> .format("jdbc")
>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>> .option("fetchsize",100)
>>> .option("dbtable", "Person").load().show(10,false)
>>>
>>>
>>> Not Working requires a CONFIG file which is example-default.xml:-
>>>
>>> val igniteDF = spark.read
>>> .format(FORMAT_IGNITE) //Data source type.
>>> .option(OPTION_TABLE, "person") //Table to read.
>>> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
>>> .load()
>>> .filter(col("id") >= 2) //Filter clause.
>>> .filter(col("name") like "%J%") //Another filter clause.
>>>
>>>
>>> Full Code:- (sparkDSLExample) function fails to connect ignite cluster
>>> which I already have
>>>
>>> package com.ignite.examples.spark
>>>
>>> import com.ignite.examples.model.Address
>>> import org.apache.ignite.{Ignite, Ignition}
>>> import org.apache.ignite.cache.query.SqlFieldsQuery
>>> import org.apache.ignite.client.{ClientCache, IgniteClient}
>>> import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
>>> import java.lang.{Long => JLong, String => JString}
>>>
>>> import org.apache.ignite.cache.query.SqlFieldsQuery
>>> import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE, OPTION_CONFIG_FILE, OPTION_TABLE}
>>> import org.apache.log4j.{Level, Logger}
>>> import org.apache.spark.sql.{SaveMode, SparkSession}
>>> import org.apache.spark.sql.functions.col
>>>
>>> object SparkClientConnectionTest {
>>>
>>> private val CACHE_NAME = "SparkCache"
>>>
>>> private val CONFIG = "/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
>>>
>>> def setupExampleData = {
>>>
>>> val cfg2 = new ClientConfiguration().setAddresses("3.88.248.113:10800")
>>> val igniteClient:IgniteClient = Ignition.startClient(cfg2)
>>>
>>> System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
>>>
>>> val cache:ClientCache[Integer, Address] = igniteClient.getOrCreateCache(CACHE_NAME)
>>>
>>> cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
>>> .setSchema("PUBLIC")).getAll
>>>
>>> cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) ) WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
>>> .setSchema("PUBLIC")).getAll
>>>
>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco", "04074").setSchema("PUBLIC")).getAll
>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road", "520003").setSchema("PUBLIC")).getAll
>>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road", "1234").setSchema("PUBLIC")).getAll
>>>
>>> System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
>>>
>>> val data=cache.query(new SqlFieldsQuery("select * from Person").setSchema("PUBLIC")).getAll
>>>
>>> println(data.toString)
>>> }
>>>
>>> def sparkDSLExample(implicit spark: SparkSession): Unit = {
>>> println("Querying using Spark DSL.")
>>> println
>>>
>>>
>>> val igniteDF = spark.read
>>> .format(FORMAT_IGNITE) //Data source type.
>>> .option(OPTION_TABLE, "person") //Table to read.
>>> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
>>> .load()
>>> .filter(col("id") >= 2) //Filter clause.
>>> .filter(col("name") like "%J%") //Another filter clause.
>>>
>>> println("Data frame schema:")
>>>
>>> igniteDF.printSchema() //Printing query schema to console.
>>>
>>> println("Data frame content:")
>>>
>>> igniteDF.show() //Printing query results to console.
>>> }
>>>
>>>
>>> def main(args: Array[String]): Unit = {
>>>
>>> setupExampleData
>>>
>>> //Creating spark session.
>>> implicit val spark = SparkSession.builder()
>>> .appName("Spark Ignite data sources example")
>>> .master("local")
>>> .config("spark.executor.instances", "2")
>>> .getOrCreate()
>>>
>>> // Adjust the logger to exclude the logs of no interest.
>>> Logger.getRootLogger.setLevel(Level.ERROR)
>>> Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
>>>
>>> //sparkDSLExample
>>>
>>>
>>> val df = spark.read
>>> .format("jdbc")
>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>> .option("fetchsize",100)
>>> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
>>> .option("dbtable", "Person").load()
>>>
>>> df.printSchema()
>>>
>>> df.createOrReplaceTempView("test")
>>>
>>> spark.sql("select * from test where id=1").show(10)
>>>
>>> spark.sql("select 4,'blah',124232").show(10)
>>>
>>> import java.sql.DriverManager
>>> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>>>
>>> import java.util.Properties
>>> val connectionProperties = new Properties()
>>>
>>> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>>>
>>> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
>>> "Person",connectionProperties)
>>>
>>> spark.read
>>> .format("jdbc")
>>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>>> .option("fetchsize",100)
>>> .option("dbtable", "Person").load().show(10,false)
>>>
>>> }
>>>
>>> }
>>>
>>>
>>> example-default.xml:-
>>>
>>> <?xml version="1.0" encoding="UTF-8"?>
>>>
>>> <!--
>>> Licensed to the Apache Software Foundation (ASF) under one or more
>>> contributor license agreements. See the NOTICE file distributed with
>>> this work for additional information regarding copyright ownership.
>>> The ASF licenses this file to You under the Apache License, Version 2.0
>>> (the "License"); you may not use this file except in compliance with
>>> the License. You may obtain a copy of the License at
>>>
>>> http://www.apache.org/licenses/LICENSE-2.0
>>>
>>> Unless required by applicable law or agreed to in writing, software
>>> distributed under the License is distributed on an "AS IS" BASIS,
>>> WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
>>> See the License for the specific language governing permissions and
>>> limitations under the License.
>>> -->
>>>
>>> <!--
>>> Ignite configuration with all defaults and enabled p2p deployment and enabled events.
>>> -->
>>> <beans xmlns="http://www.springframework.org/schema/beans"
>>> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
>>> xmlns:util="http://www.springframework.org/schema/util"
>>> xsi:schemaLocation="
>>> http://www.springframework.org/schema/beans
>>> http://www.springframework.org/schema/beans/spring-beans.xsd
>>> http://www.springframework.org/schema/util
>>> http://www.springframework.org/schema/util/spring-util.xsd">
>>> <bean abstract="true" id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
>>> <!-- Set to true to enable distributed class loading for examples, default is false. -->
>>> <property name="peerClassLoadingEnabled" value="true"/>
>>>
>>> <!-- Enable task execution events for examples. -->
>>> <property name="includeEventTypes">
>>> <list>
>>> <!--Task execution events-->
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
>>>
>>> <!--Cache events-->
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
>>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
>>> </list>
>>> </property>
>>>
>>> <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
>>> <property name="discoverySpi">
>>> <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
>>> <property name="ipFinder">
>>> <!--
>>> Ignite provides several options for automatic discovery that can be used
>>> instead os static IP based discovery. For information on all options refer
>>> to our documentation: http://apacheignite.readme.io/docs/cluster-config
>>> -->
>>> <!-- Uncomment static IP finder to enable static-based discovery of initial nodes. -->
>>> <!--<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
>>> <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
>>> <property name="addresses">
>>> <list>
>>> <!-- In distributed environment, replace with actual host IP address. -->
>>> <value>3.88.248.113:10800</value>
>>> </list>
>>> </property>
>>> </bean>
>>> </property>
>>> </bean>
>>> </property>
>>> </bean>
>>> </beans>
>>>
>>>
>>>
>>>
>>> --
>>> Thanks & Regards
>>> Sri Tummala
>>>
>>>
>>>
>>>
>>
>> --
>> Thanks & Regards
>> Sri Tummala
>>
>>
>
> --
> Thanks & Regards
> Sri Tummala
>
>
--
Thanks & Regards
Sri Tummala
Re: Apache Spark + Ignite Connection Issue
Posted by sri hari kali charan Tummala <ka...@gmail.com>.
do you mean communication ports 47100-47200 as mentioned (
https://www.gridgain.com/docs/8.7.6//installation-guide/manual-install-on-ec2)
? bare in mind I am running my spark job outside ignite ec2 box (MAC PC).
which option is right in my default.xml file?
Option 1:-
<property name="addresses">
<list>
<!-- In distributed environment,
replace with actual host IP address. -->
<value>3.88.248.113:4
<http://3.88.248.113:10800/>7100..47200</value>
</list>
</property>
Option 2:-
<property name="addresses">
<list>
<!-- In distributed environment,
replace with actual host IP address. -->
<value>3.88.248.113:4
<http://3.88.248.113:10800/>7500..47600</value>
</list>
</property>
Option 3:-
<property name="addresses">
<list>
<!-- In distributed environment,
replace with actual host IP address. -->
<value>3.88.248.113
<http://3.88.248.113:10800/></value>
</list>
</property>
Thanks
On Fri, Oct 18, 2019 at 9:18 AM sri hari kali charan Tummala <
kali.tummala@gmail.com> wrote:
> Hi Stephen ,
>
> do you mean 3.88.248.113: <http://3.88.248.113:10800/>47500..47700
> something like this? or just public ip 3.88.248.113
> <http://3.88.248.113:10800/> I tried all the possibilities none of them
> are getting connected.
>
> Thanks
> Sri
>
> On Fri, Oct 18, 2019 at 6:02 AM Stephen Darlington <
> stephen.darlington@gridgain.com> wrote:
>
>> You’re trying to connect a thick client (the Spark integration) to the
>> thin client port (10800). Your example-default.xml file needs to have the
>> same configuration as your server node(s).
>>
>> Regards,
>> Stephen
>>
>> On 17 Oct 2019, at 18:12, sri hari kali charan Tummala <
>> kali.tummala@gmail.com> wrote:
>>
>> Hi Community,
>>
>> I am trying to read and write into the Ignite cluster using apache-spark
>> I am able to do that using JDBC thin client but not native method as
>> mentioned in several spark + ignite examples.
>>
>> Right now all the spark + ignite examples launch a local ignite cluster
>> but I want my code connecting to already existing cluster (client).
>>
>> Question:-
>> *How to pass Ignite connection ip and port (10800) 10800 in
>> example-default.xml ?*
>>
>> Error:-
>> *TcpDiscoverySpi: Failed to connect to any address from IP finder (will
>> retry to join topology every 2000 ms; change 'reconnectDelay' to configure
>> the frequency of retries): [/3.88.248.113:10800
>> <http://3.88.248.113:10800/>]*
>>
>> Working (Spark + Ignite using JDBC):-
>>
>> val df = spark.read
>> .format("jdbc")
>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>> .option("fetchsize",100)
>> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
>> .option("dbtable", "Person").load()
>>
>> df.printSchema()
>>
>> df.createOrReplaceTempView("test")
>>
>> spark.sql("select * from test where id=1").show(10)
>>
>> spark.sql("select 4,'blah',124232").show(10)
>>
>> import java.sql.DriverManager
>> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>>
>> import java.util.Properties
>> val connectionProperties = new Properties()
>>
>> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>>
>> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
>> "Person",connectionProperties)
>>
>> spark.read
>> .format("jdbc")
>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>> .option("fetchsize",100)
>> .option("dbtable", "Person").load().show(10,false)
>>
>>
>> Not Working requires a CONFIG file which is example-default.xml:-
>>
>> val igniteDF = spark.read
>> .format(FORMAT_IGNITE) //Data source type.
>> .option(OPTION_TABLE, "person") //Table to read.
>> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
>> .load()
>> .filter(col("id") >= 2) //Filter clause.
>> .filter(col("name") like "%J%") //Another filter clause.
>>
>>
>> Full Code:- (sparkDSLExample) function fails to connect ignite cluster
>> which I already have
>>
>> package com.ignite.examples.spark
>>
>> import com.ignite.examples.model.Address
>> import org.apache.ignite.{Ignite, Ignition}
>> import org.apache.ignite.cache.query.SqlFieldsQuery
>> import org.apache.ignite.client.{ClientCache, IgniteClient}
>> import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
>> import java.lang.{Long => JLong, String => JString}
>>
>> import org.apache.ignite.cache.query.SqlFieldsQuery
>> import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE, OPTION_CONFIG_FILE, OPTION_TABLE}
>> import org.apache.log4j.{Level, Logger}
>> import org.apache.spark.sql.{SaveMode, SparkSession}
>> import org.apache.spark.sql.functions.col
>>
>> object SparkClientConnectionTest {
>>
>> private val CACHE_NAME = "SparkCache"
>>
>> private val CONFIG = "/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
>>
>> def setupExampleData = {
>>
>> val cfg2 = new ClientConfiguration().setAddresses("3.88.248.113:10800")
>> val igniteClient:IgniteClient = Ignition.startClient(cfg2)
>>
>> System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
>>
>> val cache:ClientCache[Integer, Address] = igniteClient.getOrCreateCache(CACHE_NAME)
>>
>> cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
>> .setSchema("PUBLIC")).getAll
>>
>> cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) ) WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
>> .setSchema("PUBLIC")).getAll
>>
>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco", "04074").setSchema("PUBLIC")).getAll
>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road", "520003").setSchema("PUBLIC")).getAll
>> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road", "1234").setSchema("PUBLIC")).getAll
>>
>> System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
>>
>> val data=cache.query(new SqlFieldsQuery("select * from Person").setSchema("PUBLIC")).getAll
>>
>> println(data.toString)
>> }
>>
>> def sparkDSLExample(implicit spark: SparkSession): Unit = {
>> println("Querying using Spark DSL.")
>> println
>>
>>
>> val igniteDF = spark.read
>> .format(FORMAT_IGNITE) //Data source type.
>> .option(OPTION_TABLE, "person") //Table to read.
>> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
>> .load()
>> .filter(col("id") >= 2) //Filter clause.
>> .filter(col("name") like "%J%") //Another filter clause.
>>
>> println("Data frame schema:")
>>
>> igniteDF.printSchema() //Printing query schema to console.
>>
>> println("Data frame content:")
>>
>> igniteDF.show() //Printing query results to console.
>> }
>>
>>
>> def main(args: Array[String]): Unit = {
>>
>> setupExampleData
>>
>> //Creating spark session.
>> implicit val spark = SparkSession.builder()
>> .appName("Spark Ignite data sources example")
>> .master("local")
>> .config("spark.executor.instances", "2")
>> .getOrCreate()
>>
>> // Adjust the logger to exclude the logs of no interest.
>> Logger.getRootLogger.setLevel(Level.ERROR)
>> Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
>>
>> //sparkDSLExample
>>
>>
>> val df = spark.read
>> .format("jdbc")
>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>> .option("fetchsize",100)
>> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
>> .option("dbtable", "Person").load()
>>
>> df.printSchema()
>>
>> df.createOrReplaceTempView("test")
>>
>> spark.sql("select * from test where id=1").show(10)
>>
>> spark.sql("select 4,'blah',124232").show(10)
>>
>> import java.sql.DriverManager
>> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>>
>> import java.util.Properties
>> val connectionProperties = new Properties()
>>
>> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>>
>> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
>> "Person",connectionProperties)
>>
>> spark.read
>> .format("jdbc")
>> .option("url", "jdbc:ignite:thin://3.88.248.113")
>> .option("fetchsize",100)
>> .option("dbtable", "Person").load().show(10,false)
>>
>> }
>>
>> }
>>
>>
>> example-default.xml:-
>>
>> <?xml version="1.0" encoding="UTF-8"?>
>>
>> <!--
>> Licensed to the Apache Software Foundation (ASF) under one or more
>> contributor license agreements. See the NOTICE file distributed with
>> this work for additional information regarding copyright ownership.
>> The ASF licenses this file to You under the Apache License, Version 2.0
>> (the "License"); you may not use this file except in compliance with
>> the License. You may obtain a copy of the License at
>>
>> http://www.apache.org/licenses/LICENSE-2.0
>>
>> Unless required by applicable law or agreed to in writing, software
>> distributed under the License is distributed on an "AS IS" BASIS,
>> WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
>> See the License for the specific language governing permissions and
>> limitations under the License.
>> -->
>>
>> <!--
>> Ignite configuration with all defaults and enabled p2p deployment and enabled events.
>> -->
>> <beans xmlns="http://www.springframework.org/schema/beans"
>> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
>> xmlns:util="http://www.springframework.org/schema/util"
>> xsi:schemaLocation="
>> http://www.springframework.org/schema/beans
>> http://www.springframework.org/schema/beans/spring-beans.xsd
>> http://www.springframework.org/schema/util
>> http://www.springframework.org/schema/util/spring-util.xsd">
>> <bean abstract="true" id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
>> <!-- Set to true to enable distributed class loading for examples, default is false. -->
>> <property name="peerClassLoadingEnabled" value="true"/>
>>
>> <!-- Enable task execution events for examples. -->
>> <property name="includeEventTypes">
>> <list>
>> <!--Task execution events-->
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
>>
>> <!--Cache events-->
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
>> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
>> </list>
>> </property>
>>
>> <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
>> <property name="discoverySpi">
>> <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
>> <property name="ipFinder">
>> <!--
>> Ignite provides several options for automatic discovery that can be used
>> instead os static IP based discovery. For information on all options refer
>> to our documentation: http://apacheignite.readme.io/docs/cluster-config
>> -->
>> <!-- Uncomment static IP finder to enable static-based discovery of initial nodes. -->
>> <!--<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
>> <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
>> <property name="addresses">
>> <list>
>> <!-- In distributed environment, replace with actual host IP address. -->
>> <value>3.88.248.113:10800</value>
>> </list>
>> </property>
>> </bean>
>> </property>
>> </bean>
>> </property>
>> </bean>
>> </beans>
>>
>>
>>
>>
>> --
>> Thanks & Regards
>> Sri Tummala
>>
>>
>>
>>
>
> --
> Thanks & Regards
> Sri Tummala
>
>
--
Thanks & Regards
Sri Tummala
Re: Apache Spark + Ignite Connection Issue
Posted by sri hari kali charan Tummala <ka...@gmail.com>.
Hi Stephen ,
do you mean 3.88.248.113: <http://3.88.248.113:10800/>47500..47700
something like this? or just public ip 3.88.248.113
<http://3.88.248.113:10800/> I tried all the possibilities none of them are
getting connected.
Thanks
Sri
On Fri, Oct 18, 2019 at 6:02 AM Stephen Darlington <
stephen.darlington@gridgain.com> wrote:
> You’re trying to connect a thick client (the Spark integration) to the
> thin client port (10800). Your example-default.xml file needs to have the
> same configuration as your server node(s).
>
> Regards,
> Stephen
>
> On 17 Oct 2019, at 18:12, sri hari kali charan Tummala <
> kali.tummala@gmail.com> wrote:
>
> Hi Community,
>
> I am trying to read and write into the Ignite cluster using apache-spark I
> am able to do that using JDBC thin client but not native method as
> mentioned in several spark + ignite examples.
>
> Right now all the spark + ignite examples launch a local ignite cluster
> but I want my code connecting to already existing cluster (client).
>
> Question:-
> *How to pass Ignite connection ip and port (10800) 10800 in
> example-default.xml ?*
>
> Error:-
> *TcpDiscoverySpi: Failed to connect to any address from IP finder (will
> retry to join topology every 2000 ms; change 'reconnectDelay' to configure
> the frequency of retries): [/3.88.248.113:10800
> <http://3.88.248.113:10800/>]*
>
> Working (Spark + Ignite using JDBC):-
>
> val df = spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://3.88.248.113")
> .option("fetchsize",100)
> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
> .option("dbtable", "Person").load()
>
> df.printSchema()
>
> df.createOrReplaceTempView("test")
>
> spark.sql("select * from test where id=1").show(10)
>
> spark.sql("select 4,'blah',124232").show(10)
>
> import java.sql.DriverManager
> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>
> import java.util.Properties
> val connectionProperties = new Properties()
>
> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>
> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
> "Person",connectionProperties)
>
> spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://3.88.248.113")
> .option("fetchsize",100)
> .option("dbtable", "Person").load().show(10,false)
>
>
> Not Working requires a CONFIG file which is example-default.xml:-
>
> val igniteDF = spark.read
> .format(FORMAT_IGNITE) //Data source type.
> .option(OPTION_TABLE, "person") //Table to read.
> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
> .load()
> .filter(col("id") >= 2) //Filter clause.
> .filter(col("name") like "%J%") //Another filter clause.
>
>
> Full Code:- (sparkDSLExample) function fails to connect ignite cluster
> which I already have
>
> package com.ignite.examples.spark
>
> import com.ignite.examples.model.Address
> import org.apache.ignite.{Ignite, Ignition}
> import org.apache.ignite.cache.query.SqlFieldsQuery
> import org.apache.ignite.client.{ClientCache, IgniteClient}
> import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
> import java.lang.{Long => JLong, String => JString}
>
> import org.apache.ignite.cache.query.SqlFieldsQuery
> import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE, OPTION_CONFIG_FILE, OPTION_TABLE}
> import org.apache.log4j.{Level, Logger}
> import org.apache.spark.sql.{SaveMode, SparkSession}
> import org.apache.spark.sql.functions.col
>
> object SparkClientConnectionTest {
>
> private val CACHE_NAME = "SparkCache"
>
> private val CONFIG = "/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
>
> def setupExampleData = {
>
> val cfg2 = new ClientConfiguration().setAddresses("3.88.248.113:10800")
> val igniteClient:IgniteClient = Ignition.startClient(cfg2)
>
> System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
>
> val cache:ClientCache[Integer, Address] = igniteClient.getOrCreateCache(CACHE_NAME)
>
> cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
> .setSchema("PUBLIC")).getAll
>
> cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) ) WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
> .setSchema("PUBLIC")).getAll
>
> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco", "04074").setSchema("PUBLIC")).getAll
> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road", "520003").setSchema("PUBLIC")).getAll
> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road", "1234").setSchema("PUBLIC")).getAll
>
> System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
>
> val data=cache.query(new SqlFieldsQuery("select * from Person").setSchema("PUBLIC")).getAll
>
> println(data.toString)
> }
>
> def sparkDSLExample(implicit spark: SparkSession): Unit = {
> println("Querying using Spark DSL.")
> println
>
>
> val igniteDF = spark.read
> .format(FORMAT_IGNITE) //Data source type.
> .option(OPTION_TABLE, "person") //Table to read.
> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
> .load()
> .filter(col("id") >= 2) //Filter clause.
> .filter(col("name") like "%J%") //Another filter clause.
>
> println("Data frame schema:")
>
> igniteDF.printSchema() //Printing query schema to console.
>
> println("Data frame content:")
>
> igniteDF.show() //Printing query results to console.
> }
>
>
> def main(args: Array[String]): Unit = {
>
> setupExampleData
>
> //Creating spark session.
> implicit val spark = SparkSession.builder()
> .appName("Spark Ignite data sources example")
> .master("local")
> .config("spark.executor.instances", "2")
> .getOrCreate()
>
> // Adjust the logger to exclude the logs of no interest.
> Logger.getRootLogger.setLevel(Level.ERROR)
> Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
>
> //sparkDSLExample
>
>
> val df = spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://3.88.248.113")
> .option("fetchsize",100)
> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
> .option("dbtable", "Person").load()
>
> df.printSchema()
>
> df.createOrReplaceTempView("test")
>
> spark.sql("select * from test where id=1").show(10)
>
> spark.sql("select 4,'blah',124232").show(10)
>
> import java.sql.DriverManager
> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113")
>
> import java.util.Properties
> val connectionProperties = new Properties()
>
> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113")
>
> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113",
> "Person",connectionProperties)
>
> spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://3.88.248.113")
> .option("fetchsize",100)
> .option("dbtable", "Person").load().show(10,false)
>
> }
>
> }
>
>
> example-default.xml:-
>
> <?xml version="1.0" encoding="UTF-8"?>
>
> <!--
> Licensed to the Apache Software Foundation (ASF) under one or more
> contributor license agreements. See the NOTICE file distributed with
> this work for additional information regarding copyright ownership.
> The ASF licenses this file to You under the Apache License, Version 2.0
> (the "License"); you may not use this file except in compliance with
> the License. You may obtain a copy of the License at
>
> http://www.apache.org/licenses/LICENSE-2.0
>
> Unless required by applicable law or agreed to in writing, software
> distributed under the License is distributed on an "AS IS" BASIS,
> WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
> See the License for the specific language governing permissions and
> limitations under the License.
> -->
>
> <!--
> Ignite configuration with all defaults and enabled p2p deployment and enabled events.
> -->
> <beans xmlns="http://www.springframework.org/schema/beans"
> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
> xmlns:util="http://www.springframework.org/schema/util"
> xsi:schemaLocation="
> http://www.springframework.org/schema/beans
> http://www.springframework.org/schema/beans/spring-beans.xsd
> http://www.springframework.org/schema/util
> http://www.springframework.org/schema/util/spring-util.xsd">
> <bean abstract="true" id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
> <!-- Set to true to enable distributed class loading for examples, default is false. -->
> <property name="peerClassLoadingEnabled" value="true"/>
>
> <!-- Enable task execution events for examples. -->
> <property name="includeEventTypes">
> <list>
> <!--Task execution events-->
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
>
> <!--Cache events-->
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
> </list>
> </property>
>
> <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
> <property name="discoverySpi">
> <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
> <property name="ipFinder">
> <!--
> Ignite provides several options for automatic discovery that can be used
> instead os static IP based discovery. For information on all options refer
> to our documentation: http://apacheignite.readme.io/docs/cluster-config
> -->
> <!-- Uncomment static IP finder to enable static-based discovery of initial nodes. -->
> <!--<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
> <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
> <property name="addresses">
> <list>
> <!-- In distributed environment, replace with actual host IP address. -->
> <value>3.88.248.113:10800</value>
> </list>
> </property>
> </bean>
> </property>
> </bean>
> </property>
> </bean>
> </beans>
>
>
>
>
> --
> Thanks & Regards
> Sri Tummala
>
>
>
>
--
Thanks & Regards
Sri Tummala
Re: Apache Spark + Ignite Connection Issue
Posted by Stephen Darlington <st...@gridgain.com>.
You’re trying to connect a thick client (the Spark integration) to the thin client port (10800). Your example-default.xml file needs to have the same configuration as your server node(s).
Regards,
Stephen
> On 17 Oct 2019, at 18:12, sri hari kali charan Tummala <ka...@gmail.com> wrote:
>
> Hi Community,
>
> I am trying to read and write into the Ignite cluster using apache-spark I am able to do that using JDBC thin client but not native method as mentioned in several spark + ignite examples.
>
> Right now all the spark + ignite examples launch a local ignite cluster but I want my code connecting to already existing cluster (client).
>
> Question:-
> How to pass Ignite connection ip and port (10800) 10800 in example-default.xml ?
>
> Error:-
> TcpDiscoverySpi: Failed to connect to any address from IP finder (will retry to join topology every 2000 ms; change 'reconnectDelay' to configure the frequency of retries): [/3.88.248.113:10800 <http://3.88.248.113:10800/>]
>
> Working (Spark + Ignite using JDBC):-
> val df = spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://3.88.248.113 <http://3.88.248.113/>")
> .option("fetchsize",100)
> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
> .option("dbtable", "Person").load()
>
> df.printSchema()
>
> df.createOrReplaceTempView("test")
>
> spark.sql("select * from test where id=1").show(10)
>
> spark.sql("select 4,'blah',124232").show(10)
>
> import java.sql.DriverManager
> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113 <http://3.88.248.113/>")
>
> import java.util.Properties
> val connectionProperties = new Properties()
>
> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113 <http://3.88.248.113/>")
>
> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113 <http://3.88.248.113/>",
> "Person",connectionProperties)
>
> spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://3.88.248.113 <http://3.88.248.113/>")
> .option("fetchsize",100)
> .option("dbtable", "Person").load().show(10,false)
>
> Not Working requires a CONFIG file which is example-default.xml:-
> val igniteDF = spark.read
> .format(FORMAT_IGNITE) //Data source type.
> .option(OPTION_TABLE, "person") //Table to read.
> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
> .load()
> .filter(col("id") >= 2) //Filter clause.
> .filter(col("name") like "%J%") //Another filter clause.
>
> Full Code:- (sparkDSLExample) function fails to connect ignite cluster which I already have
> package com.ignite.examples.spark
>
> import com.ignite.examples.model.Address
> import org.apache.ignite.{Ignite, Ignition}
> import org.apache.ignite.cache.query.SqlFieldsQuery
> import org.apache.ignite.client.{ClientCache, IgniteClient}
> import org.apache.ignite.configuration.{CacheConfiguration, ClientConfiguration}
> import java.lang.{Long => JLong, String => JString}
>
> import org.apache.ignite.cache.query.SqlFieldsQuery
> import org.apache.ignite.spark.IgniteDataFrameSettings.{FORMAT_IGNITE, OPTION_CONFIG_FILE, OPTION_TABLE}
> import org.apache.log4j.{Level, Logger}
> import org.apache.spark.sql.{SaveMode, SparkSession}
> import org.apache.spark.sql.functions.col
>
> object SparkClientConnectionTest {
>
> private val CACHE_NAME = "SparkCache"
>
> private val CONFIG = "/Users/kalit_000/Downloads/designing-event-driven-applications-apache-kafka-ecosystem/05/demos/kafka-streams-after/ApacheIgnitePoc/src/main/scala/com/ignite/examples/config/example-ignite.xml"
>
> def setupExampleData = {
>
> val cfg2 = new ClientConfiguration().setAddresses("3.88.248.113:10800 <http://3.88.248.113:10800/>")
> val igniteClient:IgniteClient = Ignition.startClient(cfg2)
>
> System.out.format(">>> Created cache [%s].\n", CACHE_NAME)
>
> val cache:ClientCache[Integer, Address] = igniteClient.getOrCreateCache(CACHE_NAME)
>
> cache.query(new SqlFieldsQuery(String.format("DROP TABLE IF EXISTS Person"))
> .setSchema("PUBLIC")).getAll
>
> cache.query(new SqlFieldsQuery(String.format("CREATE TABLE IF NOT EXISTS Person (id LONG,street varchar, zip VARCHAR, PRIMARY KEY (id) ) WITH \"VALUE_TYPE=%s\"", classOf[Address].getName))
> .setSchema("PUBLIC")).getAll
>
> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(1L.asInstanceOf[JLong],"Jameco", "04074").setSchema("PUBLIC")).getAll
> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(2L.asInstanceOf[JLong],"Bremar road", "520003").setSchema("PUBLIC")).getAll
> cache.query(new SqlFieldsQuery("INSERT INTO Person(id,street, zip) VALUES(?,?, ?)").setArgs(3L.asInstanceOf[JLong],"orange road", "1234").setSchema("PUBLIC")).getAll
>
> System.out.format(">>> Data Inserted into Cache [%s].\n", CACHE_NAME)
>
> val data=cache.query(new SqlFieldsQuery("select * from Person").setSchema("PUBLIC")).getAll
>
> println(data.toString)
> }
>
> def sparkDSLExample(implicit spark: SparkSession): Unit = {
> println("Querying using Spark DSL.")
> println
>
>
> val igniteDF = spark.read
> .format(FORMAT_IGNITE) //Data source type.
> .option(OPTION_TABLE, "person") //Table to read.
> .option(OPTION_CONFIG_FILE, CONFIG) //Ignite config.
> .load()
> .filter(col("id") >= 2) //Filter clause.
> .filter(col("name") like "%J%") //Another filter clause.
>
> println("Data frame schema:")
>
> igniteDF.printSchema() //Printing query schema to console.
>
> println("Data frame content:")
>
> igniteDF.show() //Printing query results to console.
> }
>
>
> def main(args: Array[String]): Unit = {
>
> setupExampleData
>
> //Creating spark session.
> implicit val spark = SparkSession.builder()
> .appName("Spark Ignite data sources example")
> .master("local")
> .config("spark.executor.instances", "2")
> .getOrCreate()
>
> // Adjust the logger to exclude the logs of no interest.
> Logger.getRootLogger.setLevel(Level.ERROR)
> Logger.getLogger("org.apache.ignite").setLevel(Level.INFO)
>
> //sparkDSLExample
>
>
> val df = spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://3.88.248.113 <http://3.88.248.113/>")
> .option("fetchsize",100)
> //.option("driver", "org.apache.ignite.IgniteJdbcDriver")
> .option("dbtable", "Person").load()
>
> df.printSchema()
>
> df.createOrReplaceTempView("test")
>
> spark.sql("select * from test where id=1").show(10)
>
> spark.sql("select 4,'blah',124232").show(10)
>
> import java.sql.DriverManager
> val connection = DriverManager.getConnection("jdbc:ignite:thin://3.88.248.113 <http://3.88.248.113/>")
>
> import java.util.Properties
> val connectionProperties = new Properties()
>
> connectionProperties.put("url", "jdbc:ignite:thin://3.88.248.113 <http://3.88.248.113/>")
>
> spark.sql("select 4 as ID,'blah' as STREET,124232 as ZIP").write.mode(SaveMode.Append).jdbc("jdbc:ignite:thin://3.88.248.113 <http://3.88.248.113/>",
> "Person",connectionProperties)
>
> spark.read
> .format("jdbc")
> .option("url", "jdbc:ignite:thin://3.88.248.113 <http://3.88.248.113/>")
> .option("fetchsize",100)
> .option("dbtable", "Person").load().show(10,false)
>
> }
>
> }
>
> example-default.xml:-
>
> <?xml version="1.0" encoding="UTF-8"?>
>
> <!--
> Licensed to the Apache Software Foundation (ASF) under one or more
> contributor license agreements. See the NOTICE file distributed with
> this work for additional information regarding copyright ownership.
> The ASF licenses this file to You under the Apache License, Version 2.0
> (the "License"); you may not use this file except in compliance with
> the License. You may obtain a copy of the License at
>
> http://www.apache.org/licenses/LICENSE-2.0 <http://www.apache.org/licenses/LICENSE-2.0>
>
> Unless required by applicable law or agreed to in writing, software
> distributed under the License is distributed on an "AS IS" BASIS,
> WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
> See the License for the specific language governing permissions and
> limitations under the License.
> -->
>
> <!--
> Ignite configuration with all defaults and enabled p2p deployment and enabled events.
> -->
> <beans xmlns="http://www.springframework.org/schema/beans <http://www.springframework.org/schema/beans>"
> xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance <http://www.w3.org/2001/XMLSchema-instance>"
> xmlns:util="http://www.springframework.org/schema/util <http://www.springframework.org/schema/util>"
> xsi:schemaLocation="
> http://www.springframework.org/schema/beans <http://www.springframework.org/schema/beans>
> http://www.springframework.org/schema/beans/spring-beans.xsd <http://www.springframework.org/schema/beans/spring-beans.xsd>
> http://www.springframework.org/schema/util <http://www.springframework.org/schema/util>
> http://www.springframework.org/schema/util/spring-util.xsd <http://www.springframework.org/schema/util/spring-util.xsd>">
> <bean abstract="true" id="ignite.cfg" class="org.apache.ignite.configuration.IgniteConfiguration">
> <!-- Set to true to enable distributed class loading for examples, default is false. -->
> <property name="peerClassLoadingEnabled" value="true"/>
>
> <!-- Enable task execution events for examples. -->
> <property name="includeEventTypes">
> <list>
> <!--Task execution events-->
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_STARTED"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FINISHED"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_FAILED"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_TIMEDOUT"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_SESSION_ATTR_SET"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_TASK_REDUCED"/>
>
> <!--Cache events-->
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ"/>
> <util:constant static-field="org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED"/>
> </list>
> </property>
>
> <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->
> <property name="discoverySpi">
> <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi">
> <property name="ipFinder">
> <!--
> Ignite provides several options for automatic discovery that can be used
> instead os static IP based discovery. For information on all options refer
> to our documentation: http://apacheignite.readme.io/docs/cluster-config <http://apacheignite.readme.io/docs/cluster-config>
> -->
> <!-- Uncomment static IP finder to enable static-based discovery of initial nodes. -->
> <!--<bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder">-->
> <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder">
> <property name="addresses">
> <list>
> <!-- In distributed environment, replace with actual host IP address. -->
> <value>3.88.248.113:10800 <http://3.88.248.113:10800/></value>
> </list>
> </property>
> </bean>
> </property>
> </bean>
> </property>
> </bean>
> </beans>
>
>
>
> --
> Thanks & Regards
> Sri Tummala
>