Pandas API on Spark is supported from Spark 3.2.0 (https://issues.apache.org/jira/browse/SPARK-34849), but the method you try to use on Dataframe was implemented later and introduced in the version 3.3.0.

It was introduced with this ticket https://issues.apache.org/jira/browse/SPARK-37337 in commit: https://github.com/apache/spark/commit/bc7d55fc1046a55df61fdb380629699e9959fcc6

Which basically makes changes to naming and depreciation/undepreciation of the methods.

### What changes were proposed in this pull request?
The PR is proposed to:

- Undeprecate (Spark)DataFrame.to_koalas 

- Deprecate (Spark)DataFrame.to_pandas_like and introduce (Spark)DataFrame.pandas_api instead.

### Why are the changes needed?
Currently, (Spark)DataFrame.to_pandas_on_spark is too long to memorize and inconvenient to call.
With the proposal of the PR, we may improve the user experience and make APIs more developer-friendly.

### Does this PR introduce _any_ user-facing change?
Yes.

(Spark)DataFrame.pandas_api is introduced.
(Spark)DataFrame.to_pandas_on_spark is deprecated.
(Spark)DataFrame.to_koalas is undeprecated.

For the Spark 3.2.1 you can check those:

type(df.to_koalas())

/databricks/spark/python/pyspark/sql/dataframe.py:2964: FutureWarning: DataFrame.to_koalas is deprecated. Use DataFrame.to_pandas_on_spark instead.
  warnings.warn(
Out[5]: pyspark.pandas.frame.DataFrame

or this one:

type(df.to_pandas_on_spark())

Out[6]: pyspark.pandas.frame.DataFrame

when you put .show() at the end, it is not a pyspark data frame anymore.

Remove it and it should work.

tx_ecommerce =tx_df.filter(tx_df["POS_Cardholder_Presence"]=="ECommerce")

tx_ecommerce.toPandas()

I faced the same problem, in my case it was because I had duplicate column names after the join.

I see you have report_date and marketplaceid in both dataframes. For each duplicated pair, you need to either drop one or both, or rename one of them.

Check your DataFrame with data.columns

It should print something like this

Index([u'regiment', u'company',  u'name',u'postTestScore'], dtype='object')

Check for hidden white spaces..Then you can rename with

data = data.rename(columns={'Number ': 'Number'})

The syntax is valid with Pandas DataFrames but that attribute doesn't exist for the PySpark created DataFrames. You can check out this link for the documentation.

Usually, the collect() method or the .rdd attribute would help you with these tasks.

You can use the following snippet to produce the desired result:

http_path = sdf.rdd.map(lambda row: row['http_path'].split('?'))
api_param_df = pd.DataFrame([[row[0], np.nan] if len(row) == 1 else row for row in http_path.collect()], columns=["api", "param"])
sdf = pd.concat([sdf.toPandas()['raw'], api_param_df], axis=1)

Note that I removed the comments to make it more readable and I've also substituted the regex with a simple split.

The syntax you are using is for a pandas DataFrame. To achieve this for a spark DataFrame, you should use the withColumn() method. This works great for a wide range of well defined DataFrame functions, but it's a little more complicated for user defined mapping functions.

General Case

In order to define a udf, you need to specify the output data type. For instance, if you wanted to apply a function my_func that returned a string, you could create a udf as follows:

import pyspark.sql.functions as f
my_udf = f.udf(my_func, StringType())

Then you can use my_udf to create a new column like:

df = df.withColumn('new_column', my_udf(f.col("some_column_name")))

Another option is to use select:

df = df.select("*", my_udf(f.col("some_column_name")).alias("new_column"))

Specific Problem

Using a udf

In your specific case, you want to use a dictionary to translate the values of your DataFrame.

Here is a way to define a udf for this purpose:

some_map_udf = f.udf(lambda x: some_map.get(x, None), IntegerType())

Notice that I used dict.get() because you want your udf to be robust to bad inputs.

df = df.withColumn('new_column', some_map_udf(f.col("some_column_name")))

Using DataFrame functions

Sometimes using a udf is unavoidable, but whenever possible, using DataFrame functions is usually preferred.

Here is one option to do the same thing without using the udf.

The trick is to iterate over the items in some_map to create a list of pyspark.sql.functions.when() functions.

some_map_func = [f.when(f.col("some_column_name") == k, v) for k, v in some_map.items()]
print(some_map_func)
#[Column<CASE WHEN (some_column_name = a) THEN 0 END>,
# Column<CASE WHEN (some_column_name = c) THEN 1 END>,
# Column<CASE WHEN (some_column_name = b) THEN 1 END>]

Now you can use pyspark.sql.functions.coalesce() inside of a select:

df = df.select("*", f.coalesce(*some_map_func).alias("some_column_name"))

This works because when() returns null by default if the condition is not met, and coalesce() will pick the first non-null value it encounters. Since the keys of the map are unique, at most one column will be non-null.

"sklearn.datasets" is a scikit package, where it contains a method load_iris().

load_iris(), by default return an object which holds data, target and other members in it. In order to get actual values you have to read the data and target content itself.

Whereas 'iris.csv', holds feature and target together.

FYI: If you set return_X_y as True in load_iris(), then you will directly get features and target.

from sklearn import datasets
data,target = datasets.load_iris(return_X_y=True)