Re: Asynchronous I/O poor performance

Posted by Arvid Heise-3 on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/Fwd-Asynchronous-I-O-poor-performance-tp36382p36434.html

Hi Mark,

Async wait operators cannot be chained to sources so the messages go through the network stack. Thus, having some latency is normal and cannot be avoided. It can be tuned though, but I don't think that this is the issue at hand as it should mostly impact latency and affect throughput less. Since external I/O calls are much more heavy weight than our internal communication, both the drop of throughput and the increase in latency are usually dwarfed by the external I/O call costs.

Please try to increase the thread pool for akka as written in my previous email and report back.

On Mon, Jul 6, 2020 at 9:44 PM Mark Zitnik <[hidden email]> wrote:

Hi Benchao,

i have run this in the code:

println(env.getConfig.getAutoWatermarkInterval)

and got 200 i do fully understand how watermarks and AsyncOperator operator works, but 

i have decided to make a simple test that should evaluate the time it takes to enter to the asyncInvoke method  and it looks that it takes about 80ms witch is longer than the time it take to get a response from my micro-service 

code below 

class AsyncDatabaseRequest extends RichAsyncFunction[String, (String, String)] {

  implicit lazy val executor: ExecutionContext = ExecutionContext.fromExecutor(Executors.directExecutor())

  /*
  implicit val actorSystem = ActorSystem.apply("test", None, None, Some(executor))
  implicit val materializer = ActorMaterializer()
  implicit val executionContext = actorSystem.dispatcher


  println(materializer.system.name)
  println("start")
  */
// redis-streaming-dev-new.xwudy5.ng.0001.use1.cache.amazonaws.com

  // redis-streaming-dev-001.xwudy5.0001.use1.cache.amazonaws.com
  var actorSystem: ActorSystem = null
  var materializer: ActorMaterializer = null
  var executionContext: ExecutionContextExecutor = null
  //var akkaHttp: HttpExt = null

  override def open(parameters: Configuration): Unit = {
    actorSystem = akka.actor.ActorSystem(UUID.randomUUID().toString, Some(ConfigFactory.load("application.conf")), None, Some(executor))
    materializer = ActorMaterializer()(actorSystem)
    executionContext = actorSystem.dispatcher
    //akkaHttp = Http(actorSystem)
  }

  override def close(): Unit = {
    actorSystem.terminate()
  }

  override def asyncInvoke(str: String, resultFuture: ResultFuture[(String, String)]): Unit = {
        val start = str.toLong
        val delta = System.currentTimeMillis() - start
        resultFuture.complete(Iterable((str, s"${delta}")))
  }
}


object Job {
  def main(args: Array[String]): Unit = {
    // set up the execution environment
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //env.enableCheckpointing(10)
    env.setParallelism(1)

    val someIntegers: DataStream[Long] = env.generateSequence(1, 100)
    //someIntegers.map { _ => System.currentTimeMillis()}.map{ s => System.currentTimeMillis()-s}.print()
    val x : DataStream[String] = someIntegers.map( _ => s"${System.currentTimeMillis()}")
    val resultStream: DataStream[(String, String)] = AsyncDataStream.unorderedWait(x, new AsyncDatabaseRequest(), 10L, TimeUnit.MILLISECONDS, 100)//.setParallelism(16)
      //AsyncDataStream.unorderedWait(data , new AsyncDatabaseRequest,3L,TimeUnit.SECONDS)
    resultStream.print()
    println(env.getConfig.getAutoWatermarkInterval)
    env.execute("Flink Scala API Skeleton")
  }
}

is this normal behavior?

On Mon, Jul 6, 2020 at 2:45 PM Benchao Li <[hidden email]> wrote:

Hi Mark,

According to your data, I think the config of AsyncOperator is OK.

There is one more config that might affect the throughput of AsyncOperator, it's watermark.

Because unordered async operator still keeps the order between watermarks, did you use

event time in your job, and if yes, what's the watermark interval in your job?

Mark Zitnik <[hidden email]> 于2020年7月5日周日 下午7:44写道：

Hi Benchao

The capacity is 100

Parallelism is 8 

Rpc req is 20ms 

Thanks

On Sun, 5 Jul 2020, 6:16 Benchao Li, <[hidden email]> wrote:

Hi Mark,

Could you give more details about your Flink job?

- the capacity of AsyncDataStream

- the parallelism of AsyncDataStream operator

- the time of per blocked rpc request

Mark Zitnik <[hidden email]> 于2020年7月5日周日 上午3:48写道：

Hi 

 

In my flink application I need to enrich data using AsyncDataStream.unorderedWait but I am getting poor perforce at the beginning I was just working with http call, but I have switched to grpc, I running on 8 core node and getting total of 3200 events per second my service that I am using is not fully utilized and can produce up to 10000 req/seq

Flink job flow 

Reading from Kafka ~> some enrichment with unoderedwait ~> map ~> write to Kafka 

Using Akkad grpc code written in scala 

Thanks

--

Best,
Benchao Li

--

Best,
Benchao Li

--