EI
.option("sampleSize", "500000")
Тоже не подбирает, даже с таким количеством.
Тоже не подбирает, даже с таким количеством.
Size: a a a
2019 July 16
AZ
там был ConflictType раньше, который адаптер выставлял если разные типы в разных документах
AZ
но он был вроде виден в printSchema()
EI
Заменил
modified на $date
EI
Как-то заэкранировать..
AZ
я попозже попробую воспроизвести, у меня как раз был стенд с монгой, какие у вас версии?
EI
mongo 4.0.4
scalaVersion := "2.11.12"
Spark 2.4.3
scalaVersion := "2.11.12"
Spark 2.4.3
EI
Как-то заэкранировать..
Накормило наллами)
2019 July 17
RI
привет! помогите разобраться. Не получается писать spark job'ой в hbase на кластере hdp 2.6.1.0. При этом сама джоба на кластере в майл облаке пишет. ошибки такие
The node /hbase is not in ZooKeeper. It should have been written by the master. Check the value configured in 'zookeeper.znode.parent'. There could be a mismatch with the one configured in the master. но перед этим есть сообщение INFO ZooKeeperRegistry: ClusterId read in ZooKeeper is nullRI
версии hbase 1.1.2.2.6.4.0-91 ( там где не пишется), 1.1.2.2.6.5.1100-53( работает запись). для записи используется shc-core-1.1.2-2.3-s_2.11-SNAPSHOT.jar
DP
А откуда вот эти 10 секунд?)
Из хайва стриминг - очень странно) гонять фулсканы ради дифов
Из хайва стриминг - очень странно) гонять фулсканы ради дифов
не важно сколько секунд, но если нам надо будет контролировать скорость как это сделать ?
нам нужно каждые Х сообщений в Kafka сделать delay
нам нужно каждые Х сообщений в Kafka сделать delay
O
не важно сколько секунд, но если нам надо будет контролировать скорость как это сделать ?
нам нужно каждые Х сообщений в Kafka сделать delay
нам нужно каждые Х сообщений в Kafka сделать delay
както так может сработать
partNum = 10;
rddWith1000eles
.coalesce(partNum)
.mapPartitions(part => {
// do send stuff
return part.size();
})
.collect()
.map(r -> {
Thread.sleep(Math.round(10000 / partNum));
})
partNum = 10;
rddWith1000eles
.coalesce(partNum)
.mapPartitions(part => {
// do send stuff
return part.size();
})
.collect()
.map(r -> {
Thread.sleep(Math.round(10000 / partNum));
})
DP
както так может сработать
partNum = 10;
rddWith1000eles
.coalesce(partNum)
.mapPartitions(part => {
// do send stuff
return part.size();
})
.collect()
.map(r -> {
Thread.sleep(Math.round(10000 / partNum));
})
partNum = 10;
rddWith1000eles
.coalesce(partNum)
.mapPartitions(part => {
// do send stuff
return part.size();
})
.collect()
.map(r -> {
Thread.sleep(Math.round(10000 / partNum));
})
да спс, что то похожее и мы написали, мб есть какие нибудь конфигурации что бы не писать велосипед
O
в чем велосипед, вам нужно реализовать логику, вы ее реализуете, хоть она и странная
O
такое ощущение что вы в ногу себе стреляете
DP
мб мб
A
Если нужно тротлить, то не проще ли это указать в кафка коннекторе?
A
Setting the max receiving rate - If the cluster resources is not large enough for the streaming application to process data as fast as it is being received, the receivers can be rate limited by setting a maximum rate limit in terms of records / sec. See the configuration parameters spark.streaming.receiver.maxRate for receivers and spark.streaming.kafka.maxRatePerPartition for Direct Kafka approach. In Spark 1.5, we have introduced a feature called backpressure that eliminate the need to set this rate limit, as Spark Streaming automatically figures out the rate limits and dynamically adjusts them if the processing conditions change. This backpressure can be enabled by setting the configuration parameter
spark.streaming.backpressure.enabled to true.
spark.streaming.backpressure.enabled to true.
A
В итоге делаешь с одной стороны размер итерации-батча 60 секунд
Следом тротлинг на нужное количество элемпнтов
Следом тротлинг на нужное количество элемпнтов