Advice on multiple streaming job
Hi All, Need advice on executing multiple streaming jobs. Problem:- We have 100's of streaming job. Every streaming job uses new port. Also, Spark automatically checks port from 4040 to 4056, post that it fails. One of the workaround, is to provide port explicitly. Is there a way to tackle this situation? or Am I missing any thing? Thanking you in advance. Regards, Dhaval Modi dhavalmod...@gmail.com
help needed in perforance improvement of spark structured streaming
Hi Community,
I have a use case where i need to call stored procedure through structured
streaming.
I am able to send kafka message and call stored procedure ,
but since foreach sink keeps on executing stored procedure per message
i want to combine all the messages in single dtaframe and then call stored
procedure at once
is it possible to do
current code
select('value cast "string",'topic)
.select('topic,concat_ws(",", 'value cast "string") as 'value1)
.groupBy('topic cast "string").count()
.coalesce(1)
.as[String]
.writeStream
.trigger(ProcessingTime("60 seconds"))
.option("checkpointLocation", checkpointUrl)
.foreach(new SimpleSqlServerSink(jdbcUrl, connectionProperties))
thanks
rohit
Spark with HBase on Spark Runtime 2.2.1
I wrote a simple program to read data from HBase, the program works find in
Cloudera backed by HDFS. The program works fine on SPARK RUNTIME 1.6 on
Cloudera. But does NOT work on EMR with Spark Runtime 2.2.1.
But getting an exception while testing data on EMR with S3.
// Spark conf
SparkConf sparkConf = new
SparkConf().setMaster("local[4]").setAppName("My App");
JavaSparkContext jsc = new JavaSparkContext(sparkConf);
// Hbase conf
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum","localhost");
conf.set("hbase.zookeeper.property.client.port","2181");
// Submit scan into hbase conf
// conf.set(TableInputFormat.SCAN,
TableMapReduceUtil.convertScanToString(scan));
conf.set(TableInputFormat.INPUT_TABLE, "mytable");
conf.set(TableInputFormat.SCAN_ROW_START, "startrow");
conf.set(TableInputFormat.SCAN_ROW_STOP, "endrow");
// Get RDD
JavaPairRDD source = jsc
.newAPIHadoopRDD(conf, TableInputFormat.class,
ImmutableBytesWritable.class, Result.class);
// Process RDD
System.out.println("&&& " + source.count());
0
down vote
favorite
I wrote a simple program to read data from HBase, the program works find in
Cloudera backed by HDFS.
But getting an exception while testing data on EMR with S3.
// Spark conf
SparkConf sparkConf = new
SparkConf().setMaster("local[4]").setAppName("My App");
JavaSparkContext jsc = new JavaSparkContext(sparkConf);
// Hbase conf
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum","localhost");
conf.set("hbase.zookeeper.property.client.port","2181");
// Submit scan into hbase conf
// conf.set(TableInputFormat.SCAN,
TableMapReduceUtil.convertScanToString(scan));
conf.set(TableInputFormat.INPUT_TABLE, "mytable");
conf.set(TableInputFormat.SCAN_ROW_START, "startrow");
conf.set(TableInputFormat.SCAN_ROW_STOP, "endrow");
// Get RDD
JavaPairRDD source = jsc
.newAPIHadoopRDD(conf, TableInputFormat.class,
ImmutableBytesWritable.class, Result.class);
// Process RDD
System.out.println("&&& " + source.count());
18/05/04 00:22:02 INFO MetricRegistries: Loaded MetricRegistries class
org.apache.hadoop.hbase.metrics.impl.MetricRegistriesImpl 18/05/04 00:22:02
ERROR TableInputFormat: java.io.IOException:
java.lang.reflect.InvocationTargetException at
org.apache.hadoop.hbase.client.ConnectionFactory.createConnection(ConnectionFactory.java:240)
Caused by: java.lang.reflect.InvocationTargetException at
sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) at
sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at
sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:423) at
org.apache.hadoop.hbase.client.ConnectionFactory.createConnection(ConnectionFactory.java:238)
Caused by: java.lang.IllegalAccessError: tried to access class
org.apache.hadoop.metrics2.lib.MetricsInfoImpl from class
org.apache.hadoop.metrics2.lib.DynamicMetricsRegistry at
org.apache.hadoop.metrics2.lib.DynamicMetricsRegistry.newGauge(DynamicMetricsRegistry.java:139)
at
org.apache.hadoop.hbase.zookeeper.MetricsZooKeeperSourceImpl.(MetricsZooKeeperSourceImpl.java:59)
at
org.apache.hadoop.hbase.zookeeper.MetricsZooKeeperSourceImpl.(MetricsZooKeeperSourceImpl.java:51)
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) at
sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at
sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:423) at
java.lang.Class.newInstance(Class.java:442) at
java.util.ServiceLoader$LazyIterator.nextService(ServiceLoader.java:380) ...
42 more
Exception in thread "main" java.io.IOException: Cannot create a record
reader because of a previous error. Please look at the previous
logs lines from the task's full log for more details. at
org.apache.hadoop.hbase.mapreduce.TableInputFormatBase.getSplits(TableInputFormatBase.java:270)
at
org.apache.hadoop.hbase.mapreduce.TableInputFormat.getSplits(TableInputFormat.java:256)
at org.apache.spark.rdd.NewHadoopRDD.getPartitions(NewHadoopRDD.scala:125)
at org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:252) at
org.apache.spark.rdd.RDD$$anonfun$partitions$2.apply(RDD.scala:250) at
scala.Option.getOrElse(Option.scala:121) at
org.apache.spark.rdd.RDD.partitions(RDD.scala:250) at
org.apache.spark.SparkContext.runJob(SparkContext.scala:2094) at
org.apache.spark.rdd.RDD.count(RDD.scala:1158) at
org.apache.spark.api.java.
Unable to Connect to Apache Phoenix From Spark
Simple Java Program to Connect to Phoenix DB:
SparkConf sparkConf = new SparkConf();
sparkConf.setAppName("Using-spark-phoenix-df");
sparkConf.setMaster("local[*]");
JavaSparkContext sc = new JavaSparkContext(sparkConf);
SQLContext sqlContext = new org.apache.spark.sql.SQLContext(sc);
Dataset fromPhx = sqlContext
.read()
.format("jdbc")
.options(
ImmutableMap.of("driver",
"org.apache.phoenix.jdbc.PhoenixDriver", "url",
"jdbc:phoenix:ZOOKEEPER_QUORUM_URL:/hbase",
"dbtable",
"STOCK")).load();
System.out.println(fromPhx.toJavaRDD().count());
spark-submit --class PhoenixToDataFrame \
--master yarn-client --deploy-mode client --executor-memory 1g \
--files /etc/hbase/conf/hbase-site.xml \
--name hbasespark --conf
"spark.executor.extraClassPath=/usr/lib/phoenix/phoenix-spark-4.13.0-HBase-1.4.jar:/usr/lib/phoenix/phoenix-core-4.13.0-HBase-1.4.jar:/usr/lib/phoenix-4.13.0-HBase-1.4-client.jar:/usr/lib/phoenix-client.jar:/usr/lib/phoenix-server.jar"
hbaseandspark-0.0.1-SNAPSHOT-jar-with-dependencies.jar
ERROR:
18/05/06 01:25:06 INFO Utils: Successfully started service 'sparkDriver' on
port 34053.
18/05/06 01:25:06 INFO SparkEnv: Registering MapOutputTracker
18/05/06 01:25:06 INFO SparkEnv: Registering BlockManagerMaster
18/05/06 01:25:06 INFO BlockManagerMasterEndpoint: Using
org.apache.spark.storage.DefaultTopologyMapper for getting topology
information
18/05/06 01:25:06 INFO BlockManagerMasterEndpoint:
BlockManagerMasterEndpoint up
18/05/06 01:25:06 INFO DiskBlockManager: Created local directory at
/mnt/tmp/blockmgr-08e5b083-e2bb-4389-8b30-29a1be08ee7e
18/05/06 01:25:06 INFO MemoryStore: MemoryStore started with capacity 414.4
MB
18/05/06 01:25:07 INFO SparkEnv: Registering OutputCommitCoordinator
18/05/06 01:25:07 INFO Utils: Successfully started service 'SparkUI' on port
4040.
18/05/06 01:25:07 INFO SparkUI: Bound SparkUI to 0.0.0.0, and started at
http://10.16.129.152:4040
18/05/06 01:25:07 INFO SparkContext: Added JAR
file:/home/ec2-user/test/target/hbaseandspark-0.0.1-SNAPSHOT-jar-with-dependencies.jar
at
spark://10.16.129.152:34053/jars/hbaseandspark-0.0.1-SNAPSHOT-jar-with-dependencies.jar
with timestamp 1525569907456
18/05/06 01:25:07 INFO Executor: Starting executor ID driver on host
localhost
18/05/06 01:25:07 INFO Utils: Successfully started service
'org.apache.spark.network.netty.NettyBlockTransferService' on port 34659.
18/05/06 01:25:07 INFO NettyBlockTransferService: Server created on
10.16.129.152:34659
18/05/06 01:25:07 INFO BlockManager: Using
org.apache.spark.storage.RandomBlockReplicationPolicy for block replication
policy
18/05/06 01:25:07 INFO BlockManagerMaster: Registering BlockManager
BlockManagerId(driver, 10.16.129.152, 34659, None)
18/05/06 01:25:07 INFO BlockManagerMasterEndpoint: Registering block manager
10.16.129.152:34659 with 414.4 MB RAM, BlockManagerId(driver, 10.16.129.152,
34659, None)
18/05/06 01:25:07 INFO BlockManagerMaster: Registered BlockManager
BlockManagerId(driver, 10.16.129.152, 34659, None)
18/05/06 01:25:07 INFO BlockManager: external shuffle service port = 7337
18/05/06 01:25:07 INFO BlockManager: Initialized BlockManager:
BlockManagerId(driver, 10.16.129.152, 34659, None)
18/05/06 01:25:09 INFO EventLoggingListener: Logging events to
hdfs:///var/log/spark/apps/local-1525569907548
18/05/06 01:25:09 INFO SharedState: loading hive config file:
file:/etc/spark/conf.dist/hive-site.xml
18/05/06 01:25:09 INFO SharedState: Setting hive.metastore.warehouse.dir
('null') to the value of spark.sql.warehouse.dir
('hdfs:///user/spark/warehouse').
18/05/06 01:25:09 INFO SharedState: Warehouse path is
'hdfs:///user/spark/warehouse'.
18/05/06 01:25:11 INFO StateStoreCoordinatorRef: Registered
StateStoreCoordinator endpoint
18/05/06 01:25:12 INFO ConnectionQueryServicesImpl: An instance of
ConnectionQueryServices was created.
18/05/06 01:25:12 WARN RpcControllerFactory: Cannot load configured
"hbase.rpc.controllerfactory.class"
(org.apache.hadoop.hbase.ipc.controller.ClientRpcControllerFactory) from
hbase-site.xml, falling back to use default RpcControllerFactory
18/05/06 01:25:12 INFO RecoverableZooKeeper: Process
identifier=hconnection-0x50110971 connecting to ZooKeeper
ensemble=ZOOKEEPER_QUORUM_URL:2181
18/05/06 01:25:13 INFO MetricsConfig: loaded properties from
hadoop-metrics2.properties
18/05/06 01:25:13 INFO MetricsSystemImpl: Scheduled Metric snapshot period
at 300 second(s).
18/05/06 01:25:13 INFO MetricsSystemImpl: HBase metrics system started
18/05/06 01:25:13 INFO MetricRegistries: Loaded MetricRegistries class
org.apache
