Implement Job Service control loop for stream ingestion jobs - #1140

This log is a bit vague? What do I as a a Feast admin understand from this?

Suggestion?

By the way, why is JOB_CACHE all caps?

It's a global constant variable (in a sense that we assign value to it once and never change the reference). Do you have a different naming convention for such variables?

So how is this client different from self.client?

Didn't want to use the same object from 2 different threads, and it didn't look like having 2 different client objects added a lot of overhead.

Would expected_job_hashes be a better name?

We try and avoid using the word metadata since its typically used when people cant think of a good name. Can we be more descriptive with this variable? job_hash_to_table_references?

How about job_hash_to_table_refs to make it little shorter?

I think this run_control_loop function can be written a little bit cleaner by pulling out some parts of these deeply nested loops into their own private methods

Just as another 2c, at least until this becomes too unwieldy, I would rather err on the side of having more logic inline rather than extracting it into helper functions here (see this )

Question: Why is _job_to_proto a method and not a function? or better yet, why not to_proto() on the class?

That wasn't written by me, but I can change it if necessary. @oavdeev any thoughts / reasons not to change this?

p.s. I don't like this becoming a public function on SparkJob because the users will see it, and it's not a publicly facing functionality. I think a private function on SparkJob would be better.

Yeah it could be a standalone function too. I didn't want to put it on SparkJob to keep it a pure interface. Besides, it is only used by JobServiceServicer.

I didn't want to put it on SparkJob to keep it a pure interface

Wouldn't it still be a pure interface? I mean we don't have to implement to_proto() on the ABC, but only the children. I don't feel super strongly either way, but I dont think it should be a method.

Can we use lower case for concepts as a rule of thumb? feature table.

Also, are we really going to raise an exception if the job doesn't exist when they run a start method? Doesn't that create a race condition?

Can you describe the race condition? The idea is that if the job doesn't already exist and the control loop is working in the background, it'll probably start this job soon. So here we don't start the spark job, hence no race condition. We would have the race condition if we tried to start the stream ingestion job here, since the control loop could also be trying to do the same at the same time.

Decided to throw an exception instead of returning response with empty job_id, because even if we return the response the job.get_status() and other methods will fail anyways, since job_id is empty. And that error will be more difficult to debug to users than the explicit exception.

@tsotnet it seems like we can take two approaches with the whole job service structure.

1. A thin wrapper on the Feast Client. Don't access internal methods directly and maintain abstraction. Don't duplicate logic in two places (inside and outside the client).
2. Provide a shared library as the backend for both the client and service. So the service doesnt have any client instances, but it creates its own config, launches its own jobs, etc. It still uses the same methods as the client, but it just doesnt call the client.

It seems like the current approach is a mix of these two. We pull in the client for config, and then we access internal methods directly. Are we concerned about the internal dependencies we are creating here?

Not sure I agree with this. The launcher layer also extensively uses client, so they aren't clearly decoupled from each other. If we really want the good decoupling, not only should job service not use client, but launcher should also not use client at all. We can do that, but not sure this PR is the right place for it.

The launcher layer also extensively uses client, so they aren't clearly decoupled from each other.

Yes. I don't think it's a good idea to have the client in the launcher.

If we really want the good decoupling, not only should job service not use client, but launcher should also not use client at all.

Well which approach do you think is preferrable? I am ok with the JS using the client, I just dont think we should mix the two approaches (use client and then bypass client).

but not sure this PR is the right place for it.

Knowing that we don't want to increase the internal dependencies (bypassing Client), can we try and not make things worse with this PR at least?

fyi for others: we discussed this in Zoom. Decided to move ensure_stream_ingestion_jobs in client and job service will call that. In a separate PR we will remove usage of client in launcher and replace it with config reader. In job service as well, we'll replace client with config object whenever possible.

Would it be better to use __hash__() on Python objects and to use a class/object of the job here instead of params?

Didn't want to reuse __hash__ which is intended for a different purpose (e.g. deduping in dictionaries & sets). The params class is more general to define its __hash__ for this purpose only.

__hash__ wouldn't really work here anyway since it would change if you restart the pod/process

Oh I thought Willem meant overriding & implementing __hash__ function instead of get_job_hash. Using existing __hash__ directly would just not work as Oleg stated.

I meant overriding the __hash__ function so that the unique (and unchanging) properties of the job are used to create a hash. That way we can have Job1 == Job1 and get away from writing custom getters/setters and storage for hash.

@tsotnet do you think this is a bad idea? I don't think it has to be in this PR, and there are other problems like params and jobs being separate, but do you think job level hashes at the __hash__ level is something we can use to simplify the code base later? It seems like we don't need to have separate storage of hash values if we have this.

I'd def prefer an explicit get_job_hash() and make it part of the params interface (like it is rn). If I was reading feast code, overriding __hash__ in this matter to do this would be surprising to me.

Please add docstrings to methods with types.

It really seems like this ensure function should be a method on the client itself and not specific to the job service.

Can you elaborate, why? It seems like this is a specific functionality to Job Service. Why should client worry about this? Also, this'll make it difficult to decouple client & job service.

ensure is the same as start all jobs necessary to be started. If we believe that starting a specific job is important (from client) then ensure is also important. In fact there is already a use case for this when you cant run stateful processes but you can launch jobs (Databricks scheduler or Github Actions launching Spark jobs on Databricks/EMR).

Also, this'll make it difficult to decouple client & job service.

How so? Wouldn't it improve the decoupling if the logic is only in the client?

fyi for others: discussed in Zoom, see above comment.

You are missing a type definition here and a complete docstring. Also, should this be a public method?

Probably shouldn't be a public method. Do we need all functions to be compliant to the generated docstrings (with parameters & return type). Sometimes it's overly verbose when the function is simple to describe.

We don't have to follow it religiously. For simple methods we can just have a sentence explaining the method. It's ok to skip in this case I guess.

Got it, I'll add a more extensive comment since I'm moving this to the client and modifying it a little.

@oavdeev @tsotnet Isn't it possible for us to simplify things to def add_job(self, job: SparkJob)? And have the id and hash be a property of the job? It seems like we are doing a lot of bookkeeping here with state that is somewhere derived from each other (id and hash from job).

Just a thought.

It's possible - I'm working on it. I'm combining this idea with what I've discussed with Oleg before. Basically, job hash should be part of StreamIngestionJob. In this case a lot of things become simpler to do - e.g. launcher won't have the ugly new list_jobs_by_hash method anymore.

run the check is a bit vague, maybe ensures stream ingestion jobs are running for all projects?

How do you see a Feast admin using this information @tsotnet?

s.t?

What the difference between SparkJobStatus & JobStatusProto?
Why can't we have only proto version used everywhere instead of SparkJobStatus and omit this mapping?

Let's keep SparkJob a pure abstract class, even if we have to duplicate a couple lines of code. This version is substantially less readable to me than the previous one, since now JobProto construction logic is spread across the class hierarchy.

@pyalex i agree seems like there no reason why we shouldn't unify those. I'd address this in a separate PR though (after all, that enum wasn't introduced or even changed in this PR).