Hi We (@Dejan Mijic) have been looking into integrating venice into our existing data processing pipeline and decided the most reliable way of doing that would be writing to kafka directly as we have the capacity to do that within exisiting spark jobs. The data is already clean and validated to have the schema at this point and it will be a batch reload not an update. So could you tell us or point us to documentation/ code snippets on. 1. First of all is it a possible solution ? 2. What would be the kafka topic/ kafka topic naming scheme, that we need to write to a particular store? 3. Is there any steps that need to be carried out on Venice api? a. before the ingestion b. after the ingestion

Hi Moditha! I think the most straightforward way would be to leverage the Venice Push Job, and all of the steps would be taken care of for you.

Hi Felix, we would agree with that, if we'd be able to use it / integrate in our data producing pipelines 🙂 .

Got it. I remember we talked about it a few months ago. Is the main blocker the Spark’s Scala version we use being different than yours?

In general the dependencies + the nature of VenicePushJob (i.e. being a "plain java app" caused integration difficulties). @Moditha Hewasinghage can provide more details there.

Hi Felix, The issue with integrating the push job is that its too tightly coupled with a fat jar and tries to create a spark cluster within the job and requires data to be written to s3/storage before pushing. In our usecase we have our own spark clusters that runs specific jars (our own). So its a matter of finding the correct balance and seems like writing to kafka will be the most straightforward way to do it as spark natively supports it, so theoretically we just need to know the extra steps needed to achieve this. Since our usecase is simple we do not need all the guards around the push job. So it felt convenient to ask here than dissecting the pusherjob on our own in case you have some docs on it.

I want to tag @Nisarg Thakkar here in case he has any ideas around this

> But if I understand correctly, you’re saying that writing your data to S3 to serve as input for VPJ is too troublesome (or costly? etc?) yes exactly. I was thinking more in the lines of using the kafka writer on spark https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html#writing-data-to-kafka our usage is quite minimal it will always be batch writes to a specific store. The schema is pre-determined and will rarely change.

Anything is possible, but the more custom stuff you do, the more brittle it will be… so I think it is preferred to stick to the official APIs as much as possible so that if we need to change implementation details you get better compatibility across upgrades.

Thanks Felix, yes more or less I was looking for the write protocol and fulfil that while pushing directly to kafka through spark. This way we only have to link to kafka at runtime with no other dependancies. We also have a backup plan of running VPJ with spark on k8s. The best would be to make VPJ work on EMR/databricks if someone has insights on that it would be great.

So… again, anything is possible. But linking just to Kafka would be quite tedious IMHO. I think at the bare minimum you would need to take a dependency on the

venice-common

module, in order to leverage

KafkaKeySerializer

,

KafkaValueSerializer

and `DefaultVenicePartitioner`… that would be the way to plug the Venice bits and pieces into Kafka. Although even that is not quite the full Venice protocol, because for example: 1. We support chunking large values across several Kafka messages, and reassembling them on the other side… but that part of the logic is handled by the

VeniceWriter

(which is also located in `venice-common`… so not an extra dependency compared to the previous option, but it is no longer the Kafka API). Technically, it is feasible to run with chunking disabled, but some of the functionalities are unsafe in combination with chunking disabled, and I think the long-term direction is probably to eventually drop support for non-chunked stores. So this would be an example where using the lower-level APIs can make future upgrades harder for you. 2. Another example is that there are some metadata around the payload, which we use for reliability (detecting transmission issues such as data loss at the Kafka level). You would need to populate those correctly, but that is also handled out of the box by the

VeniceWriter

. 3. Finally, we expect certain Kafka producer settings, such as

max.inflight.requests = 1

for ordering guarantee reasons. Again, it can be configured manually, but it’s done out of the box with the

VW

. I would say, at the very least, using the

VW

is the lowest level abstraction I would recommend… though like I said, even the

VW

is considered a private API. But at least it handles all of the important details…

I believe your usage is batch write (with atomically removing the previously written batch data). Is this understanding correct?

yes

The "online producer" should be able to serve your use-case conceptually. However, today, it doesn't support writing to new version topics, and only works for "hybrid stores" (stores that also have nearline writes). So, it would need to be evolved to support this use-case. Another solution would be to add an implementation of AbstractDataWriterSparkJob. It can accept data frames as long as the DF follows a specific format: getUserInputDataFrame You'd also need to add an implementation of InputDataInfoProvider and need to swap it out in VPJ

If an atomic swap is desired, then I think we really ought to solve whatever friction there is in using VPJ…

I think the current mode of taking an arbitrary DF as input allows us to not land on disk. But it's the schema validation in

DefaultInputDataInfoProvider

part that still looks at HDFS. It can be swapped out through custom

InputDataInfoProvider

to fetch the schema info from other places, but it's something we still haven't had a clean solution for

For spark, can’t we get the schema from the dataset itself?

VPJ does a strict schema validation of the dataset to what is registered on the backend. Spark's schema is a catalyst schema, and Venice schemas are Avro. The current code flow is: 1. Get dataset schema from input 2. Validate key schema a. Get schema of key field b. Check if key schema is the same as what is registered 3. Validate value schemas a. Get schema of value field b. Check if the value schema matches any value schema registered So, this assumes that the underlying data has the ability to provide a strict schema that would have been registered in Venice. Since Spark DF will give us a catalyst schema, and we don't know all schemas registered in Venice, there's no good way to make this work in the current flow. I've thought about this a bit in the past that we need to switch the check ordering, and make it so that we can provide a list of schemas, and find which one the data can conform too. We've not fully fleshed this idea out, though

One more thing... The current schema check is done once by the VPJ "driver" by reading the header of all input files, and then doing a (single) call to the Venice controller. Technically, it could be possible to have a mode where this logic moves out of the driver and into the tasks. Each task would be responsible for checking the schema compatibility of just its own inputs, not all of the inputs. The only gotcha is this could result in DDOS-ing the controller, because there would be 1 request per task instead of per job, and the controller is not that scalable... but the schema compat check endpoint could be moved to the router/server which are scalable...

Thank you both for the input. It will be great if we can integrate directly integrate into existing spark jobs to write a DF. We have something inhouse inspired by https://scala-slick.org/ and https://github.com/getquill where we ensure strict typing on the code similar to RDD and then translate to spark dataset code using macros and also generate catalyst schemas and conversions when needed. It might be an overkill for Venice to go on such a path.

Let me spend some time to explore this. Maybe we can make it work without a ton of changes

Hi people, is there something we could do to help develop (expose?) the necessary functionality, or is it simply not worth pursuing that path (e.g., too brittle)?

Hey folks. So, I didn't get around to building this thing fully, but there is one thing we could get a discussion started on as maybe you'll have more experience with it. VPJ has these high-level steps: 1. Pre-processing the input: a. Input validation (mostly schemas) b. Getting a schema version from Venice's registered versions to write as metadata along with the records c. Other validations ... 2. Version creation in Venice 3. Compute job to write the data to Venice 4. Polling Venice to check if data has been ingested everywhere The biggest concern is around 1a and 1b. Today, VPJ gets an explicit Avro schema from the FS and compares it with what is registered on the store, and does a fairly strict comparison in that regard. Do you have any ideas on how we could make this work for Spark Row? One idea I was thinking was we would add a function that could tell us "Can this StructType be serialized into this Avro schema?"

Thanks for chiming in Nisarg! I think it’s great to explore new/alternative integrations, but we should focus on maximizing stability and maintainability. That means being careful to integrate at “the ideal layer”, i.e., at a layer which already takes care of as much of the complexity as possible. I’m not sure in this case what the “ideal layer” is, but I think a good starting point would be to explore using the Online Producer, since it’s the simplest, with the least dependencies. Currently, the producer does take in Avro and so Nisarg’s final question is relevant. Do we need to bridge from

StructType

to

Avro

? And if so, where does that logic belong? Should the producer have Generic types, with both an Avro and another impl? Or should the producer provide only Avro, and some other lib wrap it and perform the conversion?

Summoning @Kacper Bielecki

Hi 👋

Hi Kacper! Nice to have you here, welcome! I think it’d be great to work with you towards using VPJ, as it would be a solid starting point before doing anything more advanced / lower-level. We are definitely interested in improving the ease of use and flexibility of VPJ, including supporting other input types. Whether that is a must have or not for the MVP version of your solution, IDK, but regardless, we are interested in progressing towards that.

Hi, I am currently trying to run VPJ on EMR - which is managed hadoop / spark service from AWS. We currently have Venice test setup on EKS (AWS managed Kubernetes cluster). I will share my experiences / problems / setup with EMR once I am further in testing. I have talked yesterday with our DevOps and lead of the team that will be responsible for reading from Venice: • devops have concerns about node rotation - from time to time, they need to rotate Kubernetes nodes and they need a way to move pods gracefully • our real-time engineering concern was around cold caches when the store is updated in the batch mode and switched instantly (currently, with Voldemort, we are warming up caches before doing the switch) We will probably have to learn more about Venice to solve the above issues but if you have any hints, sharing them would be very welcome. Monday works good for the sync - I don't think I will benefit much from syncing earlier.

Are you looking into moving Venice pods around? Are you using

StatefulSet

? Do you use the operator pattern?

Hi there. We're running venice server as StatefulSet. Router and controller are running as a regular deployments

Got it. Would it be possible for you to attend this week's Contributor Sync-Up? It's at 9 am PT on Monday - which should be 5 pm in your timezone. It would be great to discuss this. We are also migrating to Kubernetes at LinkedIn, so we can share some notes. My understanding of this is quite shallow, but from what I've been led to believe so far, to run a stateful system on k8s in an operable state, we need to have an operator. At LinkedIn, there is a central team that has built a generic operator, and that integrates with an application-specific "Cluster-manager". This cluster manager knows how to move data around when there are node disruptions. explains this aspect more. For Helix-based systems, we use Helix as the cluster manager. There is support in Helix to mark instances as being evicted, and then the operator can bring up a replacement node that can take over the instances that were previously held by the outgoing node, etc. However, this still does not solve the problem about slow bootstrap from Kafka. We are working internally on adding support for Blob-transfer that new replicas can bootstrap from the local state of the servers. We have been testing it with DaVinci clients so far. Once we stabilize this functionality, we're planning to add the same functionality to servers over the next quarter (but don't hold me accountable to this date, we might slip a bit).

Sure, I'll add it to my schedule. Thanks for sharing presentation. I'll have a look.

Yeah, pod migration will not help much here

Sure, I'll join the sync meeting

Great. See you there!

Hi folks, it was good to chat earlier this week. Before we forget, can we write down what were the next steps on either sides to move this forward?

From my side, I will build Vencie without Hadoop libraries and try to run it on EMR with the libraries that are provided there.

Nice! Happy to hear you found a way to make it work! I’ll let @Nisarg Thakkar comment on our current setup, but in any case, I think it’d be good to increase the documentation with what worked for you.

Our production jobs get launched using

spark-submit

. We have a class with a

main

function that we use as an entry point and spin up an object of the

VenicePushJob

class Spark jobs are a managed service at LinkedIn. The team is working on moving away from Yarn and launch jobs on k8s instead