curl -XGET 'localhost:9200/google_plus_data/_search?pretty' -H 'Content-Type: application/json' -d'

{

  "query": { "match_all": {} },

  "size": 10,

  "sort": [ 

            { "@timestamp": "desc" } ,

            { "_uuid": "asc" } 

          ]

}'

"reason" : "Fielddata is disabled on text fields by default. Set fielddata=true on [_uuid] in order to load fielddata in memory by uninverting the inverted index. Note that this can however use significant memory. Alternatively use a keyword field instead."

>>> import datetime

>>> today = datetime.datetime.now()

>>> str(today)

'2018-01-23 14:49:37.284361'

>>> repr(today)

>>> f=1.1111111111111111

>>> str(f)

'1.11111111111'

>>> repr(f)

'1.1111111111111112'

https://docs.python.org/3.7/reference/datamodel.html#object.__repr__

object.__repr__(self)

Called by the repr() built-in function to compute the “official” string representation of an object. If at all possible, this should look like a valid Python expression that could be used to recreate an object with the same value (given an appropriate environment). If this is not possible, a string of the form <...some useful description...> should be returned. The return value must be a string object. If a class defines __repr__() but not __str__(), then __repr__() is also used when an “informal” string representation of instances of that class is required.

This is typically used for debugging, so it is important that the representation is information-rich and unambiguous.

object.__str__(self)

Called by str(object) and the built-in functions format() and print() to compute the “informal” or nicely printable string representation of an object. The return value must be a string object.

This method differs from object.__repr__() in that there is no expectation that __str__() return a valid Python expression: a more convenient or concise representation can be used.

The default implementation defined by the built-in type object calls object.__repr__().

import java.sql.Timestamp

import org.apache.spark.sql.SparkSession

import org.apache.spark.sql.functions._

val inputLines = spark.readStream

 .format("socket")

 .option("host", "localhost")

 .option("port", 9999)

 .load()

val words = inputLines.as[String].flatMap(_.split(" "))

val wordCounts = words.groupBy("value").count()

val query = wordCounts.writeStream

 .outputMode("complete")

 .format("console")

query.start()

import java.sql.Timestamp

import org.apache.spark.sql.SparkSession

import org.apache.spark.sql.functions._

val inputLines = spark.readStream

.format("socket")

.option("host", "localhost")

.option("port", 9999)

.option("includeTimestamp", true)

.load()

val words = inputLines.as[(String, Timestamp)].flatMap(line =>

line._1.split(" ").map(word => (word, line._2))).toDF("word", "timestamp")

val windowedCounts = words.withWatermark("timestamp", "10 seconds")

.groupBy(window($"timestamp", "10 seconds", "10 seconds"), $"word").count().orderBy("window")

val query = windowedCounts.writeStream

.outputMode("complete")

.format("console")

.option("truncate", "false")

query.start()

Calendar calendar = Calendar.getInstance();

Date date = calendar.getTime();

SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSXXX");

sdf.setTimeZone(TimeZone.getTimeZone("Asia/Seoul"));

String text = sdf.format(date);

System.out.println(text);

// "2018-01-19T20:34:46.177+09:00"

 Error in fetch kafka.server.ReplicaFetcherThread$FetchRequest@5db63c3b (kafka.server.ReplicaFetcherThread)

java.io.IOException: Connection to 1 was disconnected before the response was read

        at kafka.utils.NetworkClientBlockingOps$$anonfun$blockingSendAndReceive$extension$1$$anonfun$apply$1.apply(NetworkClientBlockingOps.scala:115)

        at kafka.utils.NetworkClientBlockingOps$$anonfun$blockingSendAndReceive$extension$1$$anonfun$apply$1.apply(NetworkClientBlockingOps.scala:112)

        at scala.Option.foreach(Option.scala:257)

        at kafka.utils.NetworkClientBlockingOps$$anonfun$blockingSendAndReceive$extension$1.apply(NetworkClientBlockingOps.scala:112)

        at kafka.utils.NetworkClientBlockingOps$$anonfun$blockingSendAndReceive$extension$1.apply(NetworkClientBlockingOps.scala:108)

        at kafka.utils.NetworkClientBlockingOps$.recursivePoll$1(NetworkClientBlockingOps.scala:137)

        at kafka.utils.NetworkClientBlockingOps$.kafka$utils$NetworkClientBlockingOps$$pollContinuously$extension(NetworkClientBlockingOps.scala:143)

        at kafka.utils.NetworkClientBlockingOps$.blockingSendAndReceive$extension(NetworkClientBlockingOps.scala:108)

        at kafka.server.ReplicaFetcherThread.sendRequest(ReplicaFetcherThread.scala:253)

        at kafka.server.ReplicaFetcherThread.fetch(ReplicaFetcherThread.scala:238)

        at kafka.server.ReplicaFetcherThread.fetch(ReplicaFetcherThread.scala:42)

        at kafka.server.AbstractFetcherThread.processFetchRequest(AbstractFetcherThread.scala:118)

        at kafka.server.AbstractFetcherThread.doWork(AbstractFetcherThread.scala:103)

        at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)

33676:[2018-01-18 12:54:53,992] INFO Partition [__consumer_offsets,34] on broker 1: Shrinking ISR for partition [__consumer_offsets,34] from 3,2,1 to 1 (kafka.cluster.Partition)

33687:[2018-01-18 12:55:06,164] INFO Partition [__consumer_offsets,34] on broker 1: Expanding ISR for partition [__consumer_offsets,34] from 1 to 1,2 (kafka.cluster.Partition)

33689:[2018-01-18 12:55:06,262] INFO Partition [__consumer_offsets,34] on broker 1: Expanding ISR for partition [__consumer_offsets,34] from 1,2 to 1,2,3 (kafka.cluster.Partition)

33691:[2018-01-18 12:55:23,963] INFO Partition [__consumer_offsets,34] on broker 1: Shrinking ISR for partition [__consumer_offsets,34] from 1,2,3 to 1 (kafka.cluster.Partition)

scala> val a = Array(1,2,3)

a: Array[Int] = Array(1, 2, 3)

scala> a.slice(0, a.size - 1)

res15: Array[Int] = Array(1, 2)

scala> a.slice(1, a.size - 1)

res17: Array[Int] = Array(2)

scala> a.dropRight(1)

res21: Array[Int] = Array(1, 2)

scala> a.drop(1)

res24: Array[Int] = Array(2, 3)

scala> a.dropRight(0)

res20: Array[Int] = Array(1, 2, 3)

scala> a.drop(0)

res23: Array[Int] = Array(1, 2, 3)

The reason why the back pressure didn’t work out of the box with the default parameters is that we have a bolt that has a pretty long processing time, and usually takes 0.1 second to process a single message. During a peak of events, the spout was fast enough to fill the buffer of these slow bolts. When we kept the default size of 1024 messages, a bolt had to process more than 1000 messages in 0.1s each, which would add up to 100 seconds before the last message gets processed. That didn’t work well with the timeout of 30 seconds for a tuple in the topology, and it caused many messages to fail.

The main parameter we had to tune was the buffer size. Based on the assumptions that we don’t want to have too many fails, and that our topology is not latency sensitive, we agreed on the limit that a tuple shouldn’t wait more than 10 seconds in that executor receive buffer. This means we don’t want more than 100 messages in the buffer, then we can go for a buffer size of 64 or 128. As a rule of thumb, we align the value of the topology.executor.send.buffer.size with the one of the topology.executor.receive.buffer.size. For us, tuning the buffer to the adapted size was sufficient to get the backpressure to work. It throttles the spout when the bolts can’t keep up the pace.

We barely touched the watermark values, these seem to make sense only if you have specific considerations like:

• When the spout is throttling at 0.9, it’s too late, some of the tuples are still filling the buffers, lets reduce the high watermark to 0.8
• When the spout is throttled, and the messages are dropping under 0.4, the spout has some latency to fetch data and build new messages, that causes some bolts to be idle for a small moment, lets increase the low watermark to 0.5