This also highlights other areas where we need to apply this fix, e.g. OrcSink and CsvSin,
@FloWi it simply looks like this:
object JdbcDialect {
def apply(url: String): JdbcDialect = {
if (url.toLowerCase.startsWith("jdbc:oracle")) new OracleJdbcDialect()
else new GenericJdbcDialect()
}
}
@FloWi I must say you are the first user that's using Array support for JDBC - so that's great again and thanks again :+1:
@FloWi Ok maybe there's a couple of things you could do....
1) You could have Spark write out fewer part files, e.g. restrict the number of partitions using coallesce
2) It's true that there will be N number of subscriber threads for cached thread pool, but they won't all be active as this is controlled via the parallelism param on the SinkAction, which uses a fixed thread pool for the number of partitions - I could be wrong but this is how I read it?
So in theory say if you had 100 files - it would create 100 subscription threads but only
N for parallelism should ever be active.
However if this is indeed the issue we could easily change the
cached thread pool at the subscription level to a fixed thread pool - a kind of belt and braces - we would need to somehow pass down the parallelism param, e.g. Source.toDataStream(parallelism) - or something like that.
Sure, I already repartitioned the dataframe. Maybe that wasn't clear in my issue. That's how I found out that the
I like your suggestion with a fixed threadpool that can be explicitly controlled by the parallelism on the Source-side.
parallelism=1 on the Sink-side didn't have an effect on the resource usage. During my quick debugging session I found out that the threadpool contained 187 threads and the queue had 13 left. I don't know if they were all running though. Do you know how to find that out? Threaddump and see if they are RUNNING? Or can the thread be in running state and internal eel-logic makes it wait?I like your suggestion with a fixed threadpool that can be explicitly controlled by the parallelism on the Source-side.
Yes I think you are right that memory is not freed until the shutdown on
DataStreamSource has finished its last subscription thread. I'm not entirely convinced that going to a fixed thread pool will completely eradicate your memory issue - it will only delay the memory consumption.
<Sameroom> Whoops! You've exceeded your daily message limit on this Sameroom account (it will reset in 24 hours and 0 minutes). If you have too many Tubes for your budget, pause or delete some Tubes on the Manage page. If you dont have a subscription, visit pricing to upgrade to unlimited messaging.
<Sameroom> Whoops! You've exceeded your daily message limit on this Sameroom account (it will reset in 23 hours and 56 minutes). If you have too many Tubes for your budget, pause or delete some Tubes on the Manage page. If you dont have a subscription, visit pricing to upgrade to unlimited messaging.
<Sameroom> Whoops! You've exceeded your daily message limit on this Sameroom account (it will reset in 23 hours and 56 minutes). If you have too many Tubes for your budget, pause or delete some Tubes on the Manage page. If you dont have a subscription, visit pricing to upgrade to unlimited messaging.
<Sameroom> Whoops! You've exceeded your daily message limit on this Sameroom account (it will reset in 23 hours and 51 minutes). If you have too many Tubes for your budget, pause or delete some Tubes on the Manage page. If you dont have a subscription, visit pricing to upgrade to unlimited messaging.
<Sameroom> Whoops! You've exceeded your daily message limit on this Sameroom account (it will reset in 23 hours and 50 minutes). If you have too many Tubes for your budget, pause or delete some Tubes on the Manage page. If you dont have a subscription, visit pricing to upgrade to unlimited messaging.
An explanation for the high memory consumption could be that I throttled the sink with parallelism=1 and that means that many files are slowly being processed in parallel.
I'll try the approach of limiting the threads on the DataStreamSource.
I won't use val executor = Executors.newFixedThreadPool(n)since there wouldn't be a reasonable default value for n.
I'll take the code from Executors.newCachedThreadPool() --> new ThreadPoolExecutor(0, n, 60L, TimeUnit.SECONDS, new SynchronousQueue[Runnable])(). That way I can use Int.MaxValue as a default without immediately creating loads of threads.
<Sameroom> Whoops! You've exceeded your daily message limit on this Sameroom account (it will reset in 23 hours and 41 minutes). If you have too many Tubes for your budget, pause or delete some Tubes on the Manage page. If you dont have a subscription, visit pricing to upgrade to unlimited messaging.
<Sameroom> Whoops! You've exceeded your daily message limit on this Sameroom account (it will reset in 23 hours and 18 minutes). If you have too many Tubes for your budget, pause or delete some Tubes on the Manage page. If you dont have a subscription, visit pricing to upgrade to unlimited messaging.
<Sameroom> Whoops! You've exceeded your daily message limit on this Sameroom account (it will reset in 23 hours and 17 minutes). If you have too many Tubes for your budget, pause or delete some Tubes on the Manage page. If you dont have a subscription, visit pricing to upgrade to unlimited messaging.
<Sameroom> Whoops! You've exceeded your daily message limit on this Sameroom account (it will reset in 23 hours and 15 minutes). If you have too many Tubes for your budget, pause or delete some Tubes on the Manage page. If you dont have a subscription, visit pricing to upgrade to unlimited messaging.
<Sameroom> Whoops! You've exceeded your daily message limit on this Sameroom account (it will reset in 23 hours and 12 minutes). If you have too many Tubes for your budget, pause or delete some Tubes on the Manage page. If you dont have a subscription, visit pricing to upgrade to unlimited messaging.
<Sameroom> Whoops! You've exceeded your daily message limit on this Sameroom account (it will reset in 23 hours and 12 minutes). If you have too many Tubes for your budget, pause or delete some Tubes on the Manage page. If you dont have a subscription, visit pricing to upgrade to unlimited messaging.
<Sameroom> Whoops! You've exceeded your daily message limit on this Sameroom account (it will reset in 23 hours and 6 minutes). If you have too many Tubes for your budget, pause or delete some Tubes on the Manage page. If you dont have a subscription, visit pricing to upgrade to unlimited messaging.
<Sameroom> Whoops! You've exceeded your daily message limit on this Sameroom account (it will reset in 23 hours and 4 minutes). If you have too many Tubes for your budget, pause or delete some Tubes on the Manage page. If you dont have a subscription, visit pricing to upgrade to unlimited messaging.
<Sameroom> Whoops! You've exceeded your daily message limit on this Sameroom account (it will reset in 21 hours and 45 minutes). If you have too many Tubes for your budget, pause or delete some Tubes on the Manage page. If you dont have a subscription, visit pricing to upgrade to unlimited messaging.
Hi, I trying to create create a Datastream from an iterator using DataStream.fromIterator
but for that I need first to create a schema for my case class which I do by using
Unfortunately I get an error:
Is eel incapable of handling case classes with Optional fields or should I create the field differently ?
but for that I need first to create a schema for my case class which I do by using
StructType.from[T].Unfortunately I get an error:
Message: Can not map class scala.Option to FieldTypeIs eel incapable of handling case classes with Optional fields or should I create the field differently ?
