火花流调优每个批处理大小不工作的记录数？

Question

• 创建了具有3个分区的主题
• spark.streaming.kafka.MaxRatePerPartition=60000
• spark.streaming.backpressure.enabled=true
• 创建StreamingContext时将批处理持续时间设置为10秒
• 以纱线模式运行，有2个执行程序（3个分区共4个内核）

现在我如何测试它是否起作用。

我有一个制作人，一次发送60000条消息到这个主题。当我检查spark UI时，我得到以下信息：

 batch time | Input size | processing time
 10:54:30   | 17610      | 5s
 10:54:20   | 32790      | 8s
 10:54:10   | 9600       | 3s

spark.app.id  = application_1551747423133_0677

spark.app.name = KafkaCallDEV

spark.driver.cores = 2

spark.driver.extraJavaOptions   = -XX:+UseG1GC -XX:ConcGCThreads=2
-XX:InitiatingHeapOccupancyPercent=35 -Dlog4j.configuration=log4j.properties -verbose:gc

spark.driver.memory = 3g

spark.driver.port   = 33917

spark.executor.cores = 2

spark.executor.extraJavaOptions = -XX:+UseG1GC -XX:ConcGCThreads=2
-XX:InitiatingHeapOccupancyPercent=35 -Dlog4j.configuration=log4j.properties -verbose:gc

spark.executor.id   = driver

spark.executor.instances    = 2

spark.executor.memory   = 2g

spark.master    = yarn

spark.scheduler.mode    = FIFO

spark.streaming.backpressure.enabled    = true

spark.streaming.kafka.maxRatePerPartition = 60000

spark.submit.deployMode = cluster

spark.ui.filters    = org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter

spark.ui.port = 0

spark.yarn.app.container.log.dir = /data0/yarn/container-logs/application_1551747423133_0677/container_1551747423133_0677_01_000002 

logger.info(sparkSession.sparkContext.getConf.getAll.mkString("\n"))

(spark.executor.extraJavaOptions,-XX:+UseG1GC -XX:ConcGCThreads=2

-XX:InitiatingHeapOccupancyPercent=35 -Dlog4j.configuration=log4j.properties -verbose:gc) (spark.yarn.app.id,application_1551747423133_0681)

(spark.submit.deployMode,cluster)

(spark.streaming.backpressure.enabled,true)

(spark.yarn.credentials.renewalTime,1562764821939ms)

(spark.ui.filters,org.apache.hadoop.yarn.server.webproxy.amfilter.AmIpFilter)

(spark.executor.memory,2g) 

(spark.yarn.credentials.updateTime,1562769141873ms)

(spark.driver.cores,2) 

(spark.executor.id,driver)

(spark.executor.cores,2)

(spark.master,yarn)

(spark.driver.memory,3g)

(spark.sql.warehouse.dir,/user/hive/warehouse) 

(spark.ui.port,0)

(spark.driver.extraJavaOptions,-XX:+UseG1GC -XX:ConcGCThreads=2
-XX:InitiatingHeapOccupancyPercent=35 -Dlog4j.configuration=log4j.properties -verbose:gc) 

(spark.executor.instances,2)

(spark.driver.port,37375)

    org.apache.kafka.clients.consumer.ConsumerConfig:178 - ConsumerConfig values: 
        metric.reporters = []
        metadata.max.age.ms = 300000
        partition.assignment.strategy = [org.apache.kafka.clients.consumer.RangeAssignor]
        reconnect.backoff.ms = 50
        sasl.kerberos.ticket.renew.window.factor = 0.8
        max.partition.fetch.bytes = 1048576
        ssl.keystore.type = JKS
        enable.auto.commit = false
        sasl.mechanism = GSSAPI
        interceptor.classes = null
        exclude.internal.topics = true
        ssl.truststore.password = null
        client.id = 
        ssl.endpoint.identification.algorithm = null
        max.poll.records = 60000
        check.crcs = true
        request.timeout.ms = 40000
        heartbeat.interval.ms = 3000
        auto.commit.interval.ms = 5000
        receive.buffer.bytes = 65536
        ssl.truststore.type = JKS
        ssl.truststore.location = null
        ssl.keystore.password = null
        fetch.min.bytes = 1
        send.buffer.bytes = 131072
        value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
        retry.backoff.ms = 100
        ssl.secure.random.implementation = null
        sasl.kerberos.kinit.cmd = /usr/bin/kinit
        sasl.kerberos.service.name = null
        sasl.kerberos.ticket.renew.jitter = 0.05
        ssl.trustmanager.algorithm = PKIX
        ssl.key.password = null
        fetch.max.wait.ms = 500
        sasl.kerberos.min.time.before.relogin = 60000
        connections.max.idle.ms = 540000
        session.timeout.ms = 30000
        metrics.num.samples = 2
        key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
        ssl.protocol = TLS
        ssl.provider = null
        ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
        ssl.keystore.location = null
        ssl.cipher.suites = null
        security.protocol = PLAINTEXT
        ssl.keymanager.algorithm = SunX509
        metrics.sample.window.ms = 30000
        auto.offset.reset = latest

Answer 1

spark.streaming.kafka.MaxRatePerPartition=60000意味着

由属性spark.streaming.backpressure.enabled=true启用的此直接API读取每个Kafka分区的最大速率（以每秒消息为单位）

17610+32790+9600=60000即达到批量大小。

所以你就到此为止了....

为了更好地理解我的关于火花流反压的短文

结论：如果你启用了反压，不管你发送的信息的速率是多少。它将允许固定的消息速率