и шейдинг в один и тотже пакет - очень подозрителен

netty & protobuf

пробую

Caused by: shaded.io.grpc.ManagedChannelProvider$ProviderNotFoundException: No functional channel service provider found. Try adding a dependency on the grpc-okhttp or grpc-netty artifac

нужен короче еще ребейз

хмм

а кака версия сбт ассембли?

addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.5")

надо похоже META-INF/services/ менять тоже

https://github.com/sbt/sbt-assembly/issues/250

мавен юзай

):

https://maven.apache.org/plugins/maven-shade-plugin/examples/resource-transformers.html#Concatenating_Service_Entries_with_the_ServicesResourceTransformer

проблема что сервси провайдеры не оч в скале распространены

Мавен не оч(

сча попробуем мавен впиндюрить

@krivdaallstarts gprc-netty пакет забыл

а

ты окхттп юзаешь

с обоими не работает

ты окхттп юзаешь

меня любой устроит

а gprc-java есть?

такое бывает?

вообще конечно странно

да не странно

либа зашейдилась, а имена сервисов не переименовались в папочке META-INF

"io.grpc" % "grpc-netty" % com.trueaccord.scalapb.compiler.Version.grpcJavaVersion, "com.trueaccord.scalapb" %% "scalapb-runtime-grpc" % com.trueaccord.scalapb.compiler.Version.scalapbVersion, "com.trueaccord.scalapb" %% "scalapb-runtime" % com.trueaccord.scalapb.compiler.Version.scalapbVersion, "com.rabbitmq" % "amqp-client" % "4.2.2" ), assemblyShadeRules in assembly := Seq( //ShadeRule.rename("io.grpc**" -> "shaded.io.grpc@1").inAll, ShadeRule.rename("com.google**"-> "shaded.com.google@1").inAll, ShadeRule.rename("io.netty**" -> "shaded.io.netty@1").inAll),

вот так

заработало?

да

сталкивались с ошибкой org.apache.spark.scheduler.LiveListenerBus:95 - Listener EventLoggingListener threw an exception java.lang.reflect.InvocationTargetException? Стабильно возникает на стриминг джобе через неделю работы)

Делал ли кто продьюсер для реббита и спарка, были ли проблемы с тем, что эксепшоны падали, если пытаться закрыть канал

чтот помню такое

но не про спарк, а про реббита

ну у меня типа падает отправка сообщений с мессаджем, что коннекшн олреди клозед

хотя отправка до клозет произошла

настройка включена чтоб он сам себя восстанавливал если упало физическое соединение?

у меня вот тож самое было

только я не помню с чем связано

настройка включена чтоб он сам себя восстанавливал если упало физическое соединение?

идея такова что я делаю внутри форич партишн коннекцию , потом посылку сообщений, закрытие соединения

и падает именно посылка сообщений

если соединение не закрывать , все работает

если соединение не закрывать , все работает

Посылка то асинхронная

Может ты закрываешь коннект когда туда ещё пишется

ну да, скорее всего так

и как расчитать момент когда закрывать?)

на самом деле всё проще object LazyRabbit { lazy val rabbit = { val factory = new ConnectionFactory() factory.setUri(rabbitUri) factory.setVirtualHost("/") val connection = factory.newConnection val channel = connection.createChannel (connection,channel) } }

и где ты тут закрываешь?

уже не закрываю

теперь закрывать не нужно

?

а почему не надо закрывать?

а почему не надо закрывать?

потому что это стримнг джоб

не нужно закрытие )

ну тогда да

а зачем тебе в одном месте создавать channel

создавай только connection

а channel уже по месту

ps channel вроде не thread-safe

коллеги

кто работал со slinding window в спарке

?

правильно я понимаю что он сначала собирает батч в течениии окна а потом его редьюсит?

точнее даже не так

хочется не хранить исходики на за время окна

а хочется смапить во что то поменьше

и потом заредьюсить

правильно я понимаю оно так не умеет?

Чтобы смапить во что-то поменьше надо, чтобы записи лежали на одном экзекьютере. Это можно сделать кастомным партишенером, а потом через mapPartitions помапить

спасибо, но основное что хотелось так и не смог придумать как

хочется заэкстендить DStream

но видимо никак

или меня в гугле забанили?

правильно я понимаю что он сначала собирает батч в течениии окна а потом его редьюсит?

1) сделать агрегат над агрегатом - собирать мелкие агрегаты например за 5 мин, а над ними уже - агрегат за час. вроде обычный стриминг такое должен уметь. structured streaming точно не умеет, но можно обойти, если писать первые агрегаты в промежуточный внешний стрим (например, топик кафки), а потом оттуда читать их для второй агрегации 2) если вопрос не столько в месте, сколько в скорости обработки, то можно использовать incremental aggregation (про это есть здесь - https://docs.cloud.databricks.com/docs/latest/databricks_guide/07%20Spark%20Streaming/10%20Window%20Aggregations.html)

Ну кароче примерно так и сделал

И вопрос как раз в месте так как хочу написать Sliding HyperLogLog

В пейпере они используют LFPM

И вот его обновлять не понимал как

Но вроде сраслось

Надо замерить

Чо как

В пейпере они 5ln n вроде как гарантируют

По памяти

Тут есть кто-нибудь, кто занимается machine learning в Нижнем Новгороде?

Лучше спросить в группе по machine learning

все группы заспамлены этим вопросом ответьте уже кто-нибудь ему