Replies: 3 comments 4 replies
-
Can't you do
Did you check this for flatmapgroups? Finally, this pretty much explains why we didn't invest in RDD APIs: #101 (comment) |
Beta Was this translation helpful? Give feedback.
-
|
Thanks for the reply. I had found that one of the the biggest advantages of mappartitions was in the way I can easily distribute workloads for processing. I use it to distribute an operation that will handle a batch of records at a time (eg. after repartitioning in one way or another).
I will do more investigation into the UDF idea. The UDF examples I tested would only operate on a single row at a time. This is a problem if there is a lot of overhead that is needed for initializing the worker before operating on rows - like connecting to a database or creating a lookup table, or what-not. Maybe there is a pattern available for the initialization of the .net worker process, prior to the start of a task.
Yes, I had found that and intend to test it a bit. I think my main concern is the requirement of creating artificial groupings before being able to operate on batches of records. Also the groupings may be wrong-sized or imbalanced, which is would be a problem. On the other hand, mappartitions doesn't have any prerequisite grouping requirement. I'm convinced that batching rows together during processing is always preferred over handling them one at a time. There are efficiencies that can be gained by operating on a batch of rows. When using mappartition, the unit of work is typically hundreds or thousands of rows, rather than just one . Generally it just depends on the size of the dataset, and the number of worker nodes, and the arguments that were applied when repartitioning. |
Beta Was this translation helpful? Give feedback.
-
Could you share some code? You can do this in If you are doing
You can check the
You can use a static variable, but could be a problematic on Windows b/c it can create a worker process per task.
I am not entirely sure what this mean, but even though the regular UDF is invoked per row, Spark batches up the rows and send it over to CLR. If this is still a concern, you can check the VectorUDF I shared above. |
Beta Was this translation helpful? Give feedback.
-
|
Going back to the suggestion to use RelationalGroupedDataset Apply (ie. Apache Arrow RecordBatch)... This isn't working for me. I tested the example code Microsoft.Spark.Examples.Sql.Batch.VectorDataFrameUdfs... and I get a pretty scary error. CIould this be based on my environment? I'm assuming it works on Linux? |
Beta Was this translation helpful? Give feedback.
-
|
After giving up on RelationalGroupedDataset , I also started testing the VectorUdf stuff which you pointed me to. That seems promising. I'll see how far I get with that. The VectorUdf stuff appears to be idiomatically different than mapPartitions, which is why I wasn't finding this before. I shared some pseudocode with you below which shows how my methods were organized in scala. The methods receive an input dataframe and return a modified output dataframe that is basically a copy of the input, but incorporates one or more additional columns. Working with VectorUdf, and its distinct arrays of column values (Arrow Arrays) doesn't feel that natural as working with dataframes and rows. Maybe that's because it is how things are done in python and I haven't had any experience in python yet. |
Beta Was this translation helpful? Give feedback.
-
|
@imback82 You asked for me to share some sample code. Here below is an example of using mappartitions in scala to acquire a new column that has dimension keys (surrogates) for a data frame. It basically receives a dataframe and attaches a new column on it with the name DIM_ResultSurrogate. The goal is to prepare the dimension data, so that when the fact data is eventually added to the database, we can associate all the fact records to the related dimension. The DIM_ResultSurrogate is a surrogate record identifier that represents a combination of a given "InvoiceTypeCode", "InvoiceJournalizationCode", and "PostedFlag". Notice that I don't really need to care how the records are partitioned. And I don't need to do grouping. I just operate on whatever partitions are in place. I don't show the full code below (eg. how I create transformedIterable) but it should give you the general idea. You can see that within the mapPartitions delegate, I can work with a large batch of records (p_Iterator). I can also do preparation work at at the top and bottom of the delegate (represented here by opening and closing a database connection). |
Beta Was this translation helpful? Give feedback.
-
|
Hi! Are there any plans to support |
Beta Was this translation helpful? Give feedback.
-
I see that one of the topmost goals in the roadmap is to work on improving .Net support for dataframe operations.
But I don't know if I can wait. I'm pretty stuck on not having the ability to run mappartitions. This is some critical functionality that we use regularly in scala/python. We rely heavily on methods like mappartitions and flatmapgroups . These don't appear to be available in .net for spark yet, unless I'm missing something. I only found a few references to these method names in the original RDD code ... which appears to be deprecated in version 1.
As a side question, how do people use that RDD code if they really, really need it? Are people just forking the entire repo and changing all the internals to public? Or maybe they are using reflection to execute that stuff?
I did find the dataframe methods for GroupBy/Apply and these may allow me to accomplish some portion of the work ... but they aren't a perfect fit. Ideally there would be a way to send mapping operations out to any existing partitions where they sit.
Beta Was this translation helpful? Give feedback.
All reactions