[SPARK-41279] Feature parity: DataFrame API in Spark Connect

• Type: Umbrella
• Status: Open
• Priority: Critical
• Resolution: Unresolved
• Affects Version/s: 3.4.0
• Component/s: Connect
• Epic Link: Spark Connect
• Assignee: Ruifeng Zheng
• Reporter: Hyukjin Kwon
• Shepherd: Hyukjin Kwon
• Created: 28/Nov/22 01:25
• Updated: 29/Aug/23 04:15

Description

Implement DataFrame API in Spark Connect.

Issue Links

• split from: SPARK-39674 Initial protobuf definition for Spark Connect API

Sub-Tasks

• Implement DataFrame.withColumn(s): Resolved
• Support Collect() in Python client: Resolved
• Support Alias for every Relation: Resolved
• SELECT * shouldn't be empty project list in proto.: Resolved
• Refactor server side tests to only use DataFrame API: Resolved
• Initial DSL framework for protobuf testing: Resolved
• Implement `DataFrame.fillna ` and `DataFrame.na.fill `: Resolved
• Python: rename LogicalPlan.collect to LogicalPlan.to_proto: Resolved
• Input relation can be optional for Project in Connect proto: Resolved
• [Python] Implement `DataFrame.sample`: Resolved
• Support Repartition in Connect DSL: Resolved
• Implement `DataFrame.approxQuantile` and `DataFrame.stat.approxQuantile`: Resolved
• Support CreateView in Connect DSL: Resolved
• Implement `DataFrame.SelectExpr` in Python client: Resolved
• Add Deduplicate to Connect proto: Resolved
• Implement `DataFrame.summary`: Resolved
• Show detailed differences in dataframe comparison: Resolved
• Add Intersect to Connect proto and DSL: Resolved
• Reimplement df.stat.{cov, corr} with built-in sql functions: Resolved
• Reimplement `frequentItems` with dataframe operations: Resolved
• DataFrame `withColumnsRenamed` can be implemented through `RenameColumns` proto: Resolved
• Implement `DataFrame.stat.cov`: Resolved
• Add .agg() to Connect DSL: Resolved
• Add groupby to connect DSL and test more than one grouping expressions: Resolved
• Support toDF(columnNames) in Connect DSL: Resolved
• Compatible `take`, `head` and `first` API in Python client: Resolved
• Improve SET operation support in the proto and the server: Resolved
• Reimplement `crosstab` with dataframe operations: Resolved
• Implement DataFrame.CreateGlobalView in Python client: Resolved
• Implement `DataFrame.sparkSession` in Python client: Resolved
• Update relations.proto to follow Connect Proto development guidance: Resolved
• Throw exception for Collect() and recommend to use toPandas(): Resolved
• Complete Support for Except and Intersect in Python client: Resolved
• Implement `DataFrame.dropna ` and `DataFrame.na.drop `: Resolved
• Add WHERE to Connect proto and DSL: Resolved
• Add as(alias: String) to connect DSL: Resolved
• Add a dedicated logical plan for `Summary`: Resolved
• `columns` API should use `schema` API to avoid data fetching: Resolved
• Support SelectExpr which apply Projection by expressions in Strings in Connect DSL: Resolved
• Implement `DataFrame.stat.corr`: Resolved
• Implement DataFrame cross join: Resolved
• Explain API can support different modes: Resolved
• Support Join UsingColumns in proto: Resolved
• Remove `str` from Aggregate expression type: Resolved
• Implement `DataFrame.sortWithinPartitions`: Resolved
• Implement `DataFrame.show`: Resolved
• Support List[Column] for Join's on argument.: Resolved
• Add limit and offset to Connect DSL: Resolved
• Add Sample to proto and DSL: Resolved
• Implement `DataFrame.__repr__` and `DataFrame.dtypes`: Resolved
• Implement `DataFrame.isEmpty`: Resolved
• Connect Proto should carry unparsed identifiers: Resolved
• Reimplement `summary` with dataframe operations: Resolved
• Implement `DataFrame.crosstab` and `DataFrame.stat.crosstab`: Resolved
• DataFrame.to_pandas should not return optional pandas dataframe: Resolved
• Improve `on` in Join in Python client: Resolved
• Add missing `limit(n)` in DataFrame.head: Resolved
• Complete Support for Union in Python client: Resolved
• Implement `DataFrame.drop`: Resolved
• Extend support for Join Relation: Resolved
• Dataframe.transform in Python client support: Resolved
• StructType should contain a list of StructField and each field should have a name: Resolved
• AnalyzeResult should use struct for schema: Resolved
• Change default serialization from 'broken' CSV to Spark DF JSON: Resolved
• Imports more from connect proto package to avoid calling `proto.` for Connect DSL: Resolved
• Support other data type conversion in the DataTypeProtoConverter: Resolved
• Adopt `optional` keyword from proto3 which offers `hasXXX` to differentiate if a field is set or unset: Resolved
• Add ClientType to proto to indicate which client sends a request: Resolved
• Make AnalyzePlan support multiple analysis tasks: Resolved
• Removing unused code in connect: Resolved
• `DataFrame.explain` should print and return None: Resolved
• Support string sql expressions in DF.where(): Resolved
• Add missing docs for DataFrame API: Resolved
• Improve `DataFrame.count()`: Resolved
• Implement DataFrame.toDF: Resolved
• Implement DataFrame.withColumnRenamed: Resolved
• Implement `DataFrame.replace ` and `DataFrame.na.replace `: Resolved
• Add missing avg() to DF group: Resolved
• Bug in Deduplicate Python transformation: Resolved
• Improve Documentation for Take,Tail, Limit and Offset: Resolved
• Add orderBy and drop_duplicates: Resolved
• Make `Groupby.{min, max, sum, avg, mean}` compatible with PySpark: Resolved
• Implement `DataFrame.hint`: Resolved
• Implement `DataFrame.repartitionByRange`: Resolved
• DF.groupby.agg() API should be compatible: Resolved
• Support DataFrame TempView: Resolved
• Implement `DataFrame.cube`: Resolved
• Should use SQLExpression for str arguments in Projection: Resolved
• Implement DataFrame.describe: Resolved
• Implement DataFrame. colRegex: Resolved
• Implement `DataFrame.melt` and `DataFrame.unpivot`: Resolved
• Implement DataFrame.randomSplit: Resolved
• Implement DataFrame.subtract: Resolved
• Implement DataFrame.to: Resolved
• pyspark_types_to_proto_types should supports StructType.: Resolved
• Factor GroupedData out to group.py: Resolved
• implement `DataFrame.rollup`: Resolved
• Implement `GroupedData.pivot`: Resolved
• pyspark_types_to_proto_types should supports MapType: Resolved
• Implement the command logic for print and _repr_html_: Resolved
• pyspark_types_to_proto_types should supports ArrayType: Resolved
• Implement `GroupedData.{min, max, avg, sum}`: Resolved
• Support multiple arguments in groupBy.max(...): Resolved
• Support multiple arguments in groupBy.avg(...): Resolved
• Support multiple arguments in groupBy.min(...): Resolved
• Support multiple arguments in groupBy.sum(...): Resolved
• Implement `DataFrame.freqItems ` and `DataFrame.stat.freqItems `: Resolved
• Implement `DataFrame.sampleBy ` and `DataFrame.stat.sampleBy `: Resolved
• Support star in groupBy.agg(): Resolved
• groupBy(...).agg(...).sort does not actually sort the output: Resolved
• Make getitem support filter and select: Resolved
• Implement `GroupedData.mean`: Resolved
• DataFrame.join creating ambiguous column names: Resolved
• Implement Dataframe.rdd getNumPartitions: Resolved
• Fix `isnan` function: Resolved
• DataFrame windowspec functions : unresolved columns: Resolved
• DataFrame.show(): 'Column' object is not callable: Resolved
• Fix DataFrame.describe: Resolved
• DataFrame.collect() output parity with pyspark: Resolved
• DataFrame hint parameter can be str, float or int: Resolved
• `DataFrame.collect` should handle None/NaN properly: Resolved
• DataFrame.show formatting int as double: Resolved
• Implement Dataframe.sort,sortWithinPartitions Ordering: Resolved
• Fix DataFrame.sample parameters: Resolved
• DataFrame.groupBy requires all cols be Column or str: Resolved
• DataFrame.transform: Only Column or String can be used for projections: Resolved
• Implement DataFrame.explain format to be similar to PySpark: Resolved
• DataFrame dropDuplicates should throw error on non list argument: Resolved
• Throw proper errors in Dataset.to(): Resolved
• Window.rowsBetween should handle `float("-inf")` and `float("+inf")` as argument: Resolved
• Make StructType support metadata and Implement `DataFrame.withMetadata`: Resolved
• Enable the doctest for `DataFrame.hint`: Resolved
• DataFrame.createDataFrame converting int to bigint: Resolved
• Handle Function `rand() `: Resolved
• Python: connect client lost column data with pyarrow.Table.to_pylist: Resolved
• Add `DataFrame.writeTo` to the unsupported list: Resolved
• Add the unsupported list for `GroupedData`: Resolved
• Make `withMetadata` reuse the `withColumns` proto: Resolved
• Function `slice` should handle string in params: Resolved
• Fix Function `nth_value` functions output: Resolved
• `DataFrame.collect` should support nested types: Resolved
• Function `sampleby` return parity: Resolved
• `DataFrame.intersect` doctest output has different order: Resolved
• Support DataFrame hint parameter to be list: Resolved
• DataFrame.unionByName output is wrong: Resolved
• Implement DataFrame `semanticHash`: Resolved
• Better type errors when passing wrong parameters: In Progress
• Implement DataFrame.observe: Resolved
• Parity in Error types between pyspark and connect functions: Resolved
• Implement DataFrame `sameSemantics`: Resolved
• Implement DataFrame `toLocalIterator`: Resolved
• createDataFrame supports column with map type.: Resolved
• Decouple plan transformation and validation on server side: Open
• DataFrame.join: ambiguous column: Resolved
• DataFrame.createOrReplaceGlobalTempView - SparkConnectException: requirement failed: Resolved
• DataFrame.createDataFrame datatype conversion error: Resolved
• DataFrame.show() fix map printing: Resolved
• DataFrame mapfield,structlist invalid type: Resolved
• Implement DataFrame `pandas_api`: Resolved
• DataFrame `toPandas` parity in return types: Resolved
• Support StreamingQueryListener for DataFrame.observe: Resolved
• Parity in String representation of Column: Resolved
• Parity in String representation of higher_order_function's output: Resolved
• Different exception message in DataFrame.unpivot: Resolved
• Fix map_filter and map_zip_with output order: Resolved
• Factor data conversion `arrow -> rows` out to `conversion.py`: Resolved
• Make `from_arrow_schema` support nested types: Resolved
• Different result in nested lambda function: Resolved
• Failed to test ClientE2ETestSuite with maven: Resolved
• DataFrame.createTempView - SparkConnectGrpcException: requirement failed: Resolved
• Support left_outer join: Resolved
• Different exception in DataFrame.sample: Resolved
• DataFrame.drop should handle duplicated columns properly: Resolved
• Make `DataFrame.select` support `a.*`: Resolved
• Union avoid calling `output` before analysis: Resolved
• Refactor the AnalyzePlan RPC and add `session.version`: Resolved
• Implement DataFrame.registerTempTable: Resolved
• Fix toPandas to handle timezone and map types properly.: Resolved
• Implement cache, persist, unpersist, and storageLevel: Resolved
• make mapInPandas / mapInArrow support "is_barrier": Resolved
• Fix the comparison the result with Arrow optimization enabled/disabled.: Resolved
• Fix createDataFrame from pandas with map type: Resolved
• Fix the error message of createDataFrame from np.array(0): Resolved
• Fix test_createDataFrame_with_single_data_type.: Resolved
• Fix createDataFrame from pandas to respect session timezone.: Resolved
• Fix DataFrame.collect with null struct.: Resolved
• Implement eager evaluation.: Resolved
• Decouple handle command and send response on server side: Open
• Implement DataFrame.foreach: Resolved
• Implement DataFrame.foreachPartition: Resolved
• Investigate the behavior difference in self-join: Open