feat: Kickoff Transformation implementation

[HaoXuAI]

What this PR does / why we need it:

Created a Transformation interface. it still works with the current pandas_transformation, python_transformation etc.

The next step is refactor the BatchMaterializationEngine to make it works for both Materialization and Transformation.

Which issue(s) this PR fixes:

#4584
#4277 (comment)
#4696

Misc

[franciscojavierarceo]

Very nice!! 🚀🚀🚀

[franciscojavierarceo]

Probably can just do a dictionary mapping

[HaoXuAI]

sure will add that

[franciscojavierarceo]

Probably can add the singleton parameter too

[franciscojavierarceo]

Nice

[franciscojavierarceo]

Nice

[franciscojavierarceo]

Suggested change

	spark = "spark"
	SPARK = "spark"

[franciscojavierarceo]

Do you think we can scope one type of batch transformation as our MVP? If so, which one would you feel most comfortable doing?

[HaoXuAI]

Do you think we can scope one type of batch transformation as our MVP? If so, which one would you feel most comfortable doing?

yep, maybe Spark SQL. that should be easy to implement, and we can get it live soon.

[franciscojavierarceo]

Do you think we can scope one type of batch transformation as our MVP? If so, which one would you feel most comfortable doing?

yep, maybe Spark SQL. that should be easy to implement, and we can get it live soon.

LFG!

[franciscojavierarceo]

FYI @HaoXuAI a very selfish goal I have is to use a Spark batch transformation to handle scaling embedding documents.

So in an ideal world we could do something like:

def create_embeddings(partitionData):  
  tokenizer = AutoTokenizer.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")  
  model = AutoModel.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")  
  for row in partitionData:  
      document = str(row.document)  
      inputs = tokenizer(document, padding=True, truncation=True, return_tensors="pt", max_length=512)  
      result = model(**inputs)  
      embeddings = result.last_hidden_state[:, 0, :].cpu().detach().numpy()  
      lst = embeddings.flatten().tolist()  
      yield [row.id, lst, "", "{}", None]

And

embeddings = dataset_df.rdd.mapPartitions(create_embeddings)  

Used in a batch feature view and an on demand feature view (on write) to support offline and online processing of docs for RAG. The key thing here is that we'd be able to really scale RAG offline embedding and make the transition seamless to online.

This example is from the Pinecone docs here: https://docs.pinecone.io/integrations/databricks#3-create-the-vector-embeddings

[HaoXuAI]

FYI @HaoXuAI a very selfish goal I have is to use a Spark batch transformation to handling scaling embedding documents.

So in an ideal world we could do something like:

def create_embeddings(partitionData):  
  tokenizer = AutoTokenizer.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")  
  model = AutoModel.from_pretrained("sentence-transformers/all-MiniLM-L6-v2")  
  for row in partitionData:  
      document = str(row.document)  
      inputs = tokenizer(document, padding=True, truncation=True, return_tensors="pt", max_length=512)  
      result = model(**inputs)  
      embeddings = result.last_hidden_state[:, 0, :].cpu().detach().numpy()  
      lst = embeddings.flatten().tolist()  
      yield [row.id, lst, "", "{}", None]

And

embeddings = dataset_df.rdd.mapPartitions(create_embeddings)  

Used in a batch feature view and an on demand feature view (on write) to support offline and online processing of docs for RAG. The key thing here is that we'd be able to really scale RAG offline embedding and make the transition seamless to online.

This example is from the Pinecone docs here: https://docs.pinecone.io/integrations/databricks#3-create-the-vector-embeddings

This seems to be a really good example of the BatchFeatureView + transformation

[HaoXuAI]

PR diverged, cherrypick at #5181