Vector udf via regular rdd #1025
Conversation
|
@soxofaan I plan on merging this next week, feel free to review. |
|
FYI: I just merged master in feature branch to trigger another jenkins build it's still failing though |
| #pyspark.pandas.set_option('compute.max_rows', None) | ||
| id_index = columns.index("feature_index") | ||
|
|
||
| def mapTimeseriesRows(id_bands): |
There was a problem hiding this comment.
| def mapTimeseriesRows(id_bands): | |
| def map_timeseries_rows(id_bands): |
There was a problem hiding this comment.
is it also possible to give id_bands a type annotation? it's not very clear what this variable currently is
|
|
||
| processed_df = csv_df.groupby("feature_index").apply(callback).reset_index() | ||
| csv_as_list = df.rdd.map(list).map(lambda x: (x[id_index],x)).groupByKey().map(mapTimeseriesRows) |
There was a problem hiding this comment.
| csv_as_list = df.rdd.map(list).map(lambda x: (x[id_index],x)).groupByKey().map(mapTimeseriesRows) | |
| csv_as_list = df.rdd.map(list).map(lambda x: (x[id_index],x)).groupByKey().map(map_timeseries_rows) |
There was a problem hiding this comment.
I'm still figuring this out, but csv_as_list seems bit confusing, this isn't a list of csv anymore, but an RDD of dicts, right?
| # TODO: also pass feature_index to udf? | ||
| processed = udf_function(feature_data) | ||
| bands = id_bands[1] | ||
| import pandas as pd |
There was a problem hiding this comment.
this local import shouldn't be necessary right?
| feature_data.index = feature_data.index.strftime("%Y-%m-%dT%H:%M:%SZ") | ||
| #if "date" in feature_data.columns: | ||
| # feature_data = feature_data.set_index("date") | ||
| #feature_data.index = feature_data.index.strftime("%Y-%m-%dT%H:%M:%SZ") |
There was a problem hiding this comment.
can't we remove these lines of code instead of commenting out? What is the use of keeping them?
No description provided.