Repository: pandas-dev/pandas
Pull request: BUG/CLN: Decouple Series/DataFrame.transform
- [x] closes #35811
- [x] tests added / passed
- [x] passes `black pandas`
- [x] passes `git diff upstream/master -u -- "*.py" | flake8 --diff`
- [x] whatsnew entry

First step toward #35725. Currently `transform` just calls `aggregate`, and so if we are to forbid `aggregate` from transforming, these need to be decoupled. Other than the bugfix (#35811), the only other behavioral change is in the error messages.

Assuming the bugfix #35811 is the correct behavior, docs/whatsnew also needs to be updated.

I wasn't sure if tests should be marked with #35725 or perhaps this PR #. Any guidance here?

Problem statement:
BUG/QST: Series.transform with a dictionary
What is the expected output of passing a dictionary to `Series.transform`? For example:

    s = pd.Series([1, 2, 3])
    result1 = s.transform({'a': lambda x: x + 1})
    result2 = s.transform({'a': lambda x: x + 1, 'b': lambda x: x + 2})

The docs say that `dict of axis labels -> functions` is acceptable, but I can't find any example in the docs where the output is described/shown. Under the hood, `Series.transform` is just calling `Series.aggregate` which produces the following outputs for `result1` and `result2`.

````
# result1
a  0    2
   1    3
   2    4
dtype: int64

# result2
a  0    2
   1    3
   2    4
b  0    3
   1    4
   2    5
dtype: int64
````

`result1` is deemed acceptable (the length of the result equals the length of the input) and is returned, but `result2` raises; it is not a transformation.

I am wondering if a better return would be a DataFrame where the keys are the column names ('a' and 'b' in this example).

Hints from issue discussion:
I find this is a bug actually, the docs for `pd.Series.transform` is the same with `pd.DataFrame.transform` as 
```
    @doc(
        NDFrame.transform,
        klass=_shared_doc_kwargs["klass"],
        axis=_shared_doc_kwargs["axis"],
    )
```
And I get this:
```
In [3]: s = pd.Series([1, 2, 3]) 
   ...: result1 = s.transform({'a': lambda x: x + 1}) 
   ...: result2 = s.transform({'a': lambda x: x + 1, 'b': lambda x: x + 2})        
---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
<ipython-input-3-32dce75aad7a> in <module>
      1 s = pd.Series([1, 2, 3])
      2 result1 = s.transform({'a': lambda x: x + 1})
----> 3 result2 = s.transform({'a': lambda x: x + 1, 'b': lambda x: x + 2})

~/Applications/anaconda3/lib/python3.7/site-packages/pandas/core/series.py in transform(self, func, axis, *args, **kwargs)
   3910         # Validate the axis parameter
   3911         self._get_axis_number(axis)
-> 3912         return super().transform(func, *args, **kwargs)
   3913 
   3914     def apply(self, func, convert_dtype=True, args=(), **kwds):

~/Applications/anaconda3/lib/python3.7/site-packages/pandas/core/generic.py in transform(self, func, *args, **kwargs)
  10812         result = self.agg(func, *args, **kwargs)
  10813         if is_scalar(result) or len(result) != len(self):
> 10814             raise ValueError("transforms cannot produce " "aggregated results")
  10815 
  10816         return result

ValueError: transforms cannot produce aggregated results
```


@onshek - yes, agreed. The question then is, what should be the output? A dataframe with columns being the keys of the dictionary seems reasonable to me, but I can't find any documentation saying/showing this. I think that fits in well with `DataFrame.transform` where the keys subset the columns (and, thus, are the column names in the results)
@rhshadrach I think the main point is `axis`:
```
In [11]: s = pd.Series(range(3))

In [13]: s.transform([np.sqrt, np.exp], axis=0)
Out[13]:
       sqrt       exp
0  0.000000  1.000000
1  1.000000  2.718282
2  1.414214  7.389056
```
And `s.transform([np.sqrt, np.exp], axis=1)` can be designed to (but now it fails)
```
        0  0.000000  
sqrt    1  1.000000  
        2  1.414214  
        0  1.000000  
exp     1  2.718282
        2  7.389056
```

Commit message: Added PR # to tests

Patch to review:
diff --git a/pandas/core/aggregation.py b/pandas/core/aggregation.py
index 7ca68d8289bd5..b1ec7f6f84ecc 100644
--- a/pandas/core/aggregation.py
+++ b/pandas/core/aggregation.py
@@ -21,6 +21,7 @@
 from pandas._typing import AggFuncType, FrameOrSeries, Label
 
 from pandas.core.dtypes.common import is_dict_like, is_list_like
+from pandas.core.dtypes.generic import ABCDataFrame, ABCSeries
 
 from pandas.core.base import SpecificationError
 import pandas.core.common as com
@@ -384,3 +385,102 @@ def validate_func_kwargs(
     if not columns:
         raise TypeError(no_arg_message)
     return columns, func
+
+
+def transform(
+    obj: FrameOrSeries,
+    func: Union[str, List, Dict, Callable],
+    axis: int,
+    *args,
+    **kwargs,
+) -> FrameOrSeries:
+    """
+    Transform a DataFrame or Series
+
+    Parameters
+    ----------
+    obj : DataFrame or Series
+        Object to compute the transform on.
+    func : string, function, list, or dictionary
+        Function(s) to compute the transform with.
+    axis : {0 or 'index', 1 or 'columns'}
+        Axis along which the function is applied:
+
+        * 0 or 'index': apply function to each column.
+        * 1 or 'columns': apply function to each row.
+
+    Returns
+    -------
+    DataFrame or Series
+        Result of applying ``func`` along the given axis of the
+        Series or DataFrame.
+
+    Raises
+    ------
+    ValueError
+        If the transform function fails or does not transform.
+    """
+    is_series = obj.ndim == 1
+
+    if obj._get_axis_number(axis) == 1:
+        assert not is_series
+        return transform(obj.T, func, 0, *args, **kwargs).T
+
+    if isinstance(func, list):
+        if is_series:
+            func = {com.get_callable_name(v) or v: v for v in func}
+        else:
+            func = {col: func for col in obj}
+
+    if isinstance(func, dict):
+        if not is_series:
+            cols = sorted(set(func.keys()) - set(obj.columns))
+            if len(cols) > 0:
+                raise SpecificationError(f"Column(s) {cols} do not exist")
+
+        if any(isinstance(v, dict) for v in func.values()):
+            # GH 15931 - deprecation of renaming keys
+            raise SpecificationError("nested renamer is not supported")
+
+        results = {}
+        for name, how in func.items():
+            colg = obj._gotitem(name, ndim=1)
+            try:
+                results[name] = transform(colg, how, 0, *args, **kwargs)
+            except Exception as e:
+                if str(e) == "Function did not transform":
+                    raise e
+
+        # combine results
+        if len(results) == 0:
+            raise ValueError("Transform function failed")
+        from pandas.core.reshape.concat import concat
+
+        return concat(results, axis=1)
+
+    # func is either str or callable
+    try:
+        if isinstance(func, str):
+            result = obj._try_aggregate_string_function(func, *args, **kwargs)
+        else:
+            f = obj._get_cython_func(func)
+            if f and not args and not kwargs:
+                result = getattr(obj, f)()
+            else:
+                try:
+                    result = obj.apply(func, args=args, **kwargs)
+                except Exception:
+                    result = func(obj, *args, **kwargs)
+    except Exception:
+        raise ValueError("Transform function failed")
+
+    # Functions that transform may return empty Series/DataFrame
+    # when the dtype is not appropriate
+    if isinstance(result, (ABCSeries, ABCDataFrame)) and result.empty:
+        raise ValueError("Transform function failed")
+    if not isinstance(result, (ABCSeries, ABCDataFrame)) or not result.index.equals(
+        obj.index
+    ):
+        raise ValueError("Function did not transform")
+
+    return result
diff --git a/pandas/core/base.py b/pandas/core/base.py
index 1926803d8f04b..a688302b99724 100644
--- a/pandas/core/base.py
+++ b/pandas/core/base.py
@@ -4,7 +4,7 @@
 
 import builtins
 import textwrap
-from typing import Any, Dict, FrozenSet, List, Optional, Union
+from typing import Any, Callable, Dict, FrozenSet, List, Optional, Union
 
 import numpy as np
 
@@ -560,7 +560,7 @@ def _aggregate_multiple_funcs(self, arg, _axis):
                 ) from err
             return result
 
-    def _get_cython_func(self, arg: str) -> Optional[str]:
+    def _get_cython_func(self, arg: Callable) -> Optional[str]:
         """
         if we define an internal function for this argument, return it
         """
diff --git a/pandas/core/frame.py b/pandas/core/frame.py
index c48bec9b670ad..a8540d770db1e 100644
--- a/pandas/core/frame.py
+++ b/pandas/core/frame.py
@@ -116,7 +116,7 @@
 
 from pandas.core import algorithms, common as com, nanops, ops
 from pandas.core.accessor import CachedAccessor
-from pandas.core.aggregation import reconstruct_func, relabel_result
+from pandas.core.aggregation import reconstruct_func, relabel_result, transform
 from pandas.core.arrays import Categorical, ExtensionArray
 from pandas.core.arrays.datetimelike import DatetimeLikeArrayMixin as DatetimeLikeArray
 from pandas.core.arrays.sparse import SparseFrameAccessor
@@ -7463,10 +7463,9 @@ def _aggregate(self, arg, axis=0, *args, **kwargs):
         axis=_shared_doc_kwargs["axis"],
     )
     def transform(self, func, axis=0, *args, **kwargs) -> "DataFrame":
-        axis = self._get_axis_number(axis)
-        if axis == 1:
-            return self.T.transform(func, *args, **kwargs).T
-        return super().transform(func, *args, **kwargs)
+        result = transform(self, func, axis, *args, **kwargs)
+        assert isinstance(result, DataFrame)
+        return result
 
     def apply(self, func, axis=0, raw=False, result_type=None, args=(), **kwds):
         """
diff --git a/pandas/core/generic.py b/pandas/core/generic.py
index 2af323ccc1dd3..336a7263931e4 100644
--- a/pandas/core/generic.py
+++ b/pandas/core/generic.py
@@ -10729,11 +10729,7 @@ def transform(self, func, *args, **kwargs):
         1  1.000000   2.718282
         2  1.414214   7.389056
         """
-        result = self.agg(func, *args, **kwargs)
-        if is_scalar(result) or len(result) != len(self):
-            raise ValueError("transforms cannot produce aggregated results")
-
-        return result
+        raise NotImplementedError
 
     # ----------------------------------------------------------------------
     # Misc methods
diff --git a/pandas/core/series.py b/pandas/core/series.py
index 9d84ce4b9ab2e..cc4019b6d50d7 100644
--- a/pandas/core/series.py
+++ b/pandas/core/series.py
@@ -4085,9 +4085,9 @@ def aggregate(self, func=None, axis=0, *args, **kwargs):
         axis=_shared_doc_kwargs["axis"],
     )
     def transform(self, func, axis=0, *args, **kwargs):
-        # Validate the axis parameter
-        self._get_axis_number(axis)
-        return super().transform(func, *args, **kwargs)
+        from pandas.core.aggregation import transform
+
+        return transform(self, func, axis, *args, **kwargs)
 
     def apply(self, func, convert_dtype=True, args=(), **kwds):
         """
diff --git a/pandas/tests/frame/apply/test_frame_apply.py b/pandas/tests/frame/apply/test_frame_apply.py
index 5a1e448beb40f..6084396352e41 100644
--- a/pandas/tests/frame/apply/test_frame_apply.py
+++ b/pandas/tests/frame/apply/test_frame_apply.py
@@ -1,7 +1,6 @@
 from collections import OrderedDict
 from datetime import datetime
 from itertools import chain
-import operator
 import warnings
 
 import numpy as np
@@ -14,6 +13,7 @@
 import pandas._testing as tm
 from pandas.core.apply import frame_apply
 from pandas.core.base import SpecificationError
+from pandas.tests.frame.common import zip_frames
 
 
 @pytest.fixture
@@ -1058,25 +1058,6 @@ def test_consistency_for_boxed(self, box, int_frame_const_col):
         tm.assert_frame_equal(result, expected)
 
 
-def zip_frames(frames, axis=1):
-    """
-    take a list of frames, zip them together under the
-    assumption that these all have the first frames' index/columns.
-
-    Returns
-    -------
-    new_frame : DataFrame
-    """
-    if axis == 1:
-        columns = frames[0].columns
-        zipped = [f.loc[:, c] for c in columns for f in frames]
-        return pd.concat(zipped, axis=1)
-    else:
-        index = frames[0].index
-        zipped = [f.loc[i, :] for i in index for f in frames]
-        return pd.DataFrame(zipped)
-
-
 class TestDataFrameAggregate:
     def test_agg_transform(self, axis, float_frame):
         other_axis = 1 if axis in {0, "index"} else 0
@@ -1087,10 +1068,7 @@ def test_agg_transform(self, axis, float_frame):
             f_sqrt = np.sqrt(float_frame)
 
             # ufunc
-            result = float_frame.transform(np.sqrt, axis=axis)
             expected = f_sqrt.copy()
-            tm.assert_frame_equal(result, expected)
-
             result = float_frame.apply(np.sqrt, axis=axis)
             tm.assert_frame_equal(result, expected)
 
@@ -1110,9 +1088,6 @@ def test_agg_transform(self, axis, float_frame):
                 )
             tm.assert_frame_equal(result, expected)
 
-            result = float_frame.transform([np.sqrt], axis=axis)
-            tm.assert_frame_equal(result, expected)
-
             # multiple items in list
             # these are in the order as if we are applying both
             # functions per series and then concatting
@@ -1128,38 +1103,19 @@ def test_agg_transform(self, axis, float_frame):
                 )
             tm.assert_frame_equal(result, expected)
 
-            result = float_frame.transform([np.abs, "sqrt"], axis=axis)
-            tm.assert_frame_equal(result, expected)
-
     def test_transform_and_agg_err(self, axis, float_frame):
         # cannot both transform and agg
-        msg = "transforms cannot produce aggregated results"
-        with pytest.raises(ValueError, match=msg):
-            float_frame.transform(["max", "min"], axis=axis)
-
         msg = "cannot combine transform and aggregation operations"
         with pytest.raises(ValueError, match=msg):
             with np.errstate(all="ignore"):
                 float_frame.agg(["max", "sqrt"], axis=axis)
 
-        with pytest.raises(ValueError, match=msg):
-            with np.errstate(all="ignore"):
-                float_frame.transform(["max", "sqrt"], axis=axis)
-
         df = pd.DataFrame({"A": range(5), "B": 5})
 
         def f():
             with np.errstate(all="ignore"):
                 df.agg({"A": ["abs", "sum"], "B": ["mean", "max"]}, axis=axis)
 
-    @pytest.mark.parametrize("method", ["abs", "shift", "pct_change", "cumsum", "rank"])
-    def test_transform_method_name(self, method):
-        # GH 19760
-        df = pd.DataFrame({"A": [-1, 2]})
-        result = df.transform(method)
-        expected = operator.methodcaller(method)(df)
-        tm.assert_frame_equal(result, expected)
-
     def test_demo(self):
         # demonstration tests
         df = pd.DataFrame({"A": range(5), "B": 5})
diff --git a/pandas/tests/frame/apply/test_frame_transform.py b/pandas/tests/frame/apply/test_frame_transform.py
new file mode 100644
index 0000000000000..0d7533df5c0bf
--- /dev/null
+++ b/pandas/tests/frame/apply/test_frame_transform.py
@@ -0,0 +1,184 @@
+import operator
+import re
+
+import numpy as np
+import pytest
+
+from pandas import DataFrame, MultiIndex
+import pandas._testing as tm
+from pandas.core.base import SpecificationError
+from pandas.core.groupby.base import transformation_kernels
+from pandas.tests.frame.common import zip_frames
+
+
+def test_transform(axis, float_frame):
+    # GH 35964
+    other_axis = 1 if axis in {0, "index"} else 0
+
+    with np.errstate(all="ignore"):
+        f_abs = np.abs(float_frame)
+        f_sqrt = np.sqrt(float_frame)
+
+        # ufunc
+        result = float_frame.transform(np.sqrt, axis=axis)
+        expected = f_sqrt
+        tm.assert_frame_equal(result, expected)
+
+        # list-like
+        expected = f_sqrt.copy()
+        if axis in {0, "index"}:
+            expected.columns = MultiIndex.from_product([float_frame.columns, ["sqrt"]])
+        else:
+            expected.index = MultiIndex.from_product([float_frame.index, ["sqrt"]])
+        result = float_frame.transform([np.sqrt], axis=axis)
+        tm.assert_frame_equal(result, expected)
+
+        # multiple items in list
+        # these are in the order as if we are applying both
+        # functions per series and then concatting
+        expected = zip_frames([f_abs, f_sqrt], axis=other_axis)
+        if axis in {0, "index"}:
+            expected.columns = MultiIndex.from_product(
+                [float_frame.columns, ["absolute", "sqrt"]]
+            )
+        else:
+            expected.index = MultiIndex.from_product(
+                [float_frame.index, ["absolute", "sqrt"]]
+            )
+        result = float_frame.transform([np.abs, "sqrt"], axis=axis)
+        tm.assert_frame_equal(result, expected)
+
+
+@pytest.mark.parametrize("use_apply", [True, False])
+def test_transform_udf(axis, float_frame, use_apply):
+    # GH 35964
+    # transform uses UDF either via apply or passing the entire DataFrame
+    def func(x):
+        # transform is using apply iff x is not a DataFrame
+        if use_apply == isinstance(x, DataFrame):
+            # Force transform to fallback
+            raise ValueError
+        return x + 1
+
+    result = float_frame.transform(func, axis=axis)
+    expected = float_frame + 1
+    tm.assert_frame_equal(result, expected)
+
+
+@pytest.mark.parametrize("method", ["abs", "shift", "pct_change", "cumsum", "rank"])
+def test_transform_method_name(method):
+    # GH 19760
+    df = DataFrame({"A": [-1, 2]})
+    result = df.transform(method)
+    expected = operator.methodcaller(method)(df)
+    tm.assert_frame_equal(result, expected)
+
+
+def test_transform_and_agg_err(axis, float_frame):
+    # GH 35964
+    # cannot both transform and agg
+    msg = "Function did not transform"
+    with pytest.raises(ValueError, match=msg):
+        float_frame.transform(["max", "min"], axis=axis)
+
+    msg = "Function did not transform"
+    with pytest.raises(ValueError, match=msg):
+        float_frame.transform(["max", "sqrt"], axis=axis)
+
+
+def test_agg_dict_nested_renaming_depr():
+    df = DataFrame({"A": range(5), "B": 5})
+
+    # nested renaming
+    msg = r"nested renamer is not supported"
+    with pytest.raises(SpecificationError, match=msg):
+        df.transform({"A": {"foo": "min"}, "B": {"bar": "max"}})
+
+
+def test_transform_reducer_raises(all_reductions):
+    # GH 35964
+    op = all_reductions
+    df = DataFrame({"A": [1, 2, 3]})
+    msg = "Function did not transform"
+    with pytest.raises(ValueError, match=msg):
+        df.transform(op)
+    with pytest.raises(ValueError, match=msg):
+        df.transform([op])
+    with pytest.raises(ValueError, match=msg):
+        df.transform({"A": op})
+    with pytest.raises(ValueError, match=msg):
+        df.transform({"A": [op]})
+
+
+# mypy doesn't allow adding lists of different types
+# https://github.com/python/mypy/issues/5492
+@pytest.mark.parametrize("op", [*transformation_kernels, lambda x: x + 1])
+def test_transform_bad_dtype(op):
+    # GH 35964
+    df = DataFrame({"A": 3 * [object]})  # DataFrame that will fail on most transforms
+    if op in ("backfill", "shift", "pad", "bfill", "ffill"):
+        pytest.xfail("Transform function works on any datatype")
+    msg = "Transform function failed"
+    with pytest.raises(ValueError, match=msg):
+        df.transform(op)
+    with pytest.raises(ValueError, match=msg):
+        df.transform([op])
+    with pytest.raises(ValueError, match=msg):
+        df.transform({"A": op})
+    with pytest.raises(ValueError, match=msg):
+        df.transform({"A": [op]})
+
+
+@pytest.mark.parametrize("op", transformation_kernels)
+def test_transform_multi_dtypes(op):
+    # GH 35964
+    df = DataFrame({"A": ["a", "b", "c"], "B": [1, 2, 3]})
+
+    # Determine which columns op will work on
+    columns = []
+    for column in df:
+        try:
+            df[column].transform(op)
+            columns.append(column)
+        except Exception:
+            pass
+
+    if len(columns) > 0:
+        expected = df[columns].transform([op])
+        result = df.transform([op])
+        tm.assert_equal(result, expected)
+
+        expected = df[columns].transform({column: op for column in columns})
+        result = df.transform({column: op for column in columns})
+        tm.assert_equal(result, expected)
+
+        expected = df[columns].transform({column: [op] for column in columns})
+        result = df.transform({column: [op] for column in columns})
+        tm.assert_equal(result, expected)
+
+
+@pytest.mark.parametrize("use_apply", [True, False])
+def test_transform_passes_args(use_apply):
+    # GH 35964
+    # transform uses UDF either via apply or passing the entire DataFrame
+    expected_args = [1, 2]
+    expected_kwargs = {"c": 3}
+
+    def f(x, a, b, c):
+        # transform is using apply iff x is not a DataFrame
+        if use_apply == isinstance(x, DataFrame):
+            # Force transform to fallback
+            raise ValueError
+        assert [a, b] == expected_args
+        assert c == expected_kwargs["c"]
+        return x
+
+    DataFrame([1]).transform(f, 0, *expected_args, **expected_kwargs)
+
+
+def test_transform_missing_columns(axis):
+    # GH 35964
+    df = DataFrame({"A": [1, 2], "B": [3, 4]})
+    match = re.escape("Column(s) ['C'] do not exist")
+    with pytest.raises(SpecificationError, match=match):
+        df.transform({"C": "cumsum"})
diff --git a/pandas/tests/frame/common.py b/pandas/tests/frame/common.py
index 463a140972ab5..50e8c8632a525 100644
--- a/pandas/tests/frame/common.py
+++ b/pandas/tests/frame/common.py
@@ -1,3 +1,6 @@
+from pandas import DataFrame, concat
+
+
 def _check_mixed_float(df, dtype=None):
     # float16 are most likely to be upcasted to float32
     dtypes = dict(A="float32", B="float32", C="float16", D="float64")
@@ -29,3 +32,22 @@ def _check_mixed_int(df, dtype=None):
         assert df.dtypes["C"] == dtypes["C"]
     if dtypes.get("D"):
         assert df.dtypes["D"] == dtypes["D"]
+
+
+def zip_frames(frames, axis=1):
+    """
+    take a list of frames, zip them together under the
+    assumption that these all have the first frames' index/columns.
+
+    Returns
+    -------
+    new_frame : DataFrame
+    """
+    if axis == 1:
+        columns = frames[0].columns
+        zipped = [f.loc[:, c] for c in columns for f in frames]
+        return concat(zipped, axis=1)
+    else:
+        index = frames[0].index
+        zipped = [f.loc[i, :] for i in index for f in frames]
+        return DataFrame(zipped)
diff --git a/pandas/tests/series/apply/test_series_apply.py b/pandas/tests/series/apply/test_series_apply.py
index 308398642895c..827f466e23106 100644
--- a/pandas/tests/series/apply/test_series_apply.py
+++ b/pandas/tests/series/apply/test_series_apply.py
@@ -209,23 +209,17 @@ def test_transform(self, string_series):
             f_abs = np.abs(string_series)
 
             # ufunc
-            result = string_series.transform(np.sqrt)
-            expected = f_sqrt.copy()
-            tm.assert_series_equal(result, expected)
-
             result = string_series.apply(np.sqrt)
+            expected = f_sqrt.copy()
             tm.assert_series_equal(result, expected)
 
             # list-like
-            result = string_series.transform([np.sqrt])
+            result = string_series.apply([np.sqrt])
             expected = f_sqrt.to_frame().copy()
             expected.columns = ["sqrt"]
             tm.assert_frame_equal(result, expected)
 
-            result = string_series.transform([np.sqrt])
-            tm.assert_frame_equal(result, expected)
-
-            result = string_series.transform(["sqrt"])
+            result = string_series.apply(["sqrt"])
             tm.assert_frame_equal(result, expected)
 
             # multiple items in list
@@ -236,10 +230,6 @@ def test_transform(self, string_series):
             result = string_series.apply([np.sqrt, np.abs])
             tm.assert_frame_equal(result, expected)
 
-            result = string_series.transform(["sqrt", "abs"])
-            expected.columns = ["sqrt", "abs"]
-            tm.assert_frame_equal(result, expected)
-
             # dict, provide renaming
             expected = pd.concat([f_sqrt, f_abs], axis=1)
             expected.columns = ["foo", "bar"]
@@ -250,19 +240,11 @@ def test_transform(self, string_series):
 
     def test_transform_and_agg_error(self, string_series):
         # we are trying to transform with an aggregator
-        msg = "transforms cannot produce aggregated results"
-        with pytest.raises(ValueError, match=msg):
-            string_series.transform(["min", "max"])
-
         msg = "cannot combine transform and aggregation"
         with pytest.raises(ValueError, match=msg):
             with np.errstate(all="ignore"):
                 string_series.agg(["sqrt", "max"])
 
-        with pytest.raises(ValueError, match=msg):
-            with np.errstate(all="ignore"):
-                string_series.transform(["sqrt", "max"])
-
         msg = "cannot perform both aggregation and transformation"
         with pytest.raises(ValueError, match=msg):
             with np.errstate(all="ignore"):
@@ -463,14 +445,6 @@ def test_agg_cython_table_raises(self, series, func, expected):
             # e.g. Series('a b'.split()).cumprod() will raise
             series.agg(func)
 
-    def test_transform_none_to_type(self):
-        # GH34377
-        df = pd.DataFrame({"a": [None]})
-
-        msg = "DataFrame constructor called with incompatible data and dtype"
-        with pytest.raises(TypeError, match=msg):
-            df.transform({"a": int})
-
 
 class TestSeriesMap:
     def test_map(self, datetime_series):
diff --git a/pandas/tests/series/apply/test_series_transform.py b/pandas/tests/series/apply/test_series_transform.py
new file mode 100644
index 0000000000000..54d4cd0de0d6a
--- /dev/null
+++ b/pandas/tests/series/apply/test_series_transform.py
@@ -0,0 +1,154 @@
+import numpy as np
+import pytest
+
+from pandas import DataFrame, Series, concat
+import pandas._testing as tm
+from pandas.core.base import SpecificationError
+from pandas.core.groupby.base import transformation_kernels
+
+
+def test_transform(string_series):
+    # GH 35964
+    with np.errstate(all="ignore"):
+        # transforming functions
+        f_sqrt = np.sqrt(string_series)
+        f_abs = np.abs(string_series)
+
+        # ufunc
+        result = string_series.transform(np.sqrt)
+        expected = f_sqrt.copy()
+        tm.assert_series_equal(result, expected)
+
+        # list-like
+        result = string_series.transform([np.sqrt])
+        expected = f_sqrt.to_frame().copy()
+        expected.columns = ["sqrt"]
+        tm.assert_frame_equal(result, expected)
+
+        result = string_series.transform(["sqrt"])
+        tm.assert_frame_equal(result, expected)
+
+        # multiple items in list
+        # these are in the order as if we are applying both functions per
+        # series and then concatting
+        result = string_series.transform(["sqrt", "abs"])
+        expected = concat([f_sqrt, f_abs], axis=1)
+        expected.columns = ["sqrt", "abs"]
+        tm.assert_frame_equal(result, expected)
+
+        # dict, provide renaming
+        expected = concat([f_sqrt, f_abs], axis=1)
+        expected.columns = ["foo", "bar"]
+        result = string_series.transform({"foo": np.sqrt, "bar": np.abs})
+        tm.assert_frame_equal(result, expected)
+
+
+def test_transform_udf(axis, string_series):
+    # GH 35964
+    # via apply
+    def func(x):
+        if isinstance(x, Series):
+            raise ValueError
+        return x + 1
+
+    result = string_series.transform(func)
+    expected = string_series + 1
+    tm.assert_series_equal(result, expected)
+
+    # via map Series -> Series
+    def func(x):
+        if not isinstance(x, Series):
+            raise ValueError
+        return x + 1
+
+    result = string_series.transform(func)
+    expected = string_series + 1
+    tm.assert_series_equal(result, expected)
+
+
+def test_transform_wont_agg(string_series):
+    # GH 35964
+    # we are trying to transform with an aggregator
+    msg = "Function did not transform"
+    with pytest.raises(ValueError, match=msg):
+        string_series.transform(["min", "max"])
+
+    msg = "Function did not transform"
+    with pytest.raises(ValueError, match=msg):
+        with np.errstate(all="ignore"):
+            string_series.transform(["sqrt", "max"])
+
+
+def test_transform_none_to_type():
+    # GH34377
+    df = DataFrame({"a": [None]})
+    msg = "Transform function failed"
+    with pytest.raises(ValueError, match=msg):
+        df.transform({"a": int})
+
+
+def test_transform_reducer_raises(all_reductions):
+    # GH 35964
+    op = all_reductions
+    s = Series([1, 2, 3])
+    msg = "Function did not transform"
+    with pytest.raises(ValueError, match=msg):
+        s.transform(op)
+    with pytest.raises(ValueError, match=msg):
+        s.transform([op])
+    with pytest.raises(ValueError, match=msg):
+        s.transform({"A": op})
+    with pytest.raises(ValueError, match=msg):
+        s.transform({"A": [op]})
+
+
+# mypy doesn't allow adding lists of different types
+# https://github.com/python/mypy/issues/5492
+@pytest.mark.parametrize("op", [*transformation_kernels, lambda x: x + 1])
+def test_transform_bad_dtype(op):
+    # GH 35964
+    s = Series(3 * [object])  # Series that will fail on most transforms
+    if op in ("backfill", "shift", "pad", "bfill", "ffill"):
+        pytest.xfail("Transform function works on any datatype")
+    msg = "Transform function failed"
+    with pytest.raises(ValueError, match=msg):
+        s.transform(op)
+    with pytest.raises(ValueError, match=msg):
+        s.transform([op])
+    with pytest.raises(ValueError, match=msg):
+        s.transform({"A": op})
+    with pytest.raises(ValueError, match=msg):
+        s.transform({"A": [op]})
+
+
+@pytest.mark.parametrize("use_apply", [True, False])
+def test_transform_passes_args(use_apply):
+    # GH 35964
+    # transform uses UDF either via apply or passing the entire Series
+    expected_args = [1, 2]
+    expected_kwargs = {"c": 3}
+
+    def f(x, a, b, c):
+        # transform is using apply iff x is not a Series
+        if use_apply == isinstance(x, Series):
+            # Force transform to fallback
+            raise ValueError
+        assert [a, b] == expected_args
+        assert c == expected_kwargs["c"]
+        return x
+
+    Series([1]).transform(f, 0, *expected_args, **expected_kwargs)
+
+
+def test_transform_axis_1_raises():
+    # GH 35964
+    msg = "No axis named 1 for object type Series"
+    with pytest.raises(ValueError, match=msg):
+        Series([1]).transform("sum", axis=1)
+
+
+def test_transform_nested_renamer():
+    # GH 35964
+    match = "nested renamer is not supported"
+    with pytest.raises(SpecificationError, match=match):
+        Series([1]).transform({"A": {"B": ["sum"]}})


Code review comments:
diff_hunk: @@ -29,3 +32,22 @@ def _check_mixed_int(df, dtype=None):
         assert df.dtypes["C"] == dtypes["C"]
     if dtypes.get("D"):
         assert df.dtypes["D"] == dtypes["D"]
+
+
+def zip_frames(frames, axis=1):
path: pandas/tests/frame/common.py
text: @user1:
can you type args and output

diff_hunk: @@ -0,0 +1,154 @@
+import numpy as np
+import pytest
+
+from pandas import DataFrame, Series, concat
+import pandas._testing as tm
+from pandas.core.base import SpecificationError
+from pandas.core.groupby.base import transformation_kernels
+
+
+def test_transform(string_series):
+    # GH 35964
+    with np.errstate(all="ignore"):
+        # transforming functions
+        f_sqrt = np.sqrt(string_series)
+        f_abs = np.abs(string_series)
+
+        # ufunc
+        result = string_series.transform(np.sqrt)
+        expected = f_sqrt.copy()
+        tm.assert_series_equal(result, expected)
+
+        # list-like
path: pandas/tests/series/apply/test_series_transform.py
text: @user1:
can you split this one as well

diff_hunk: @@ -10729,11 +10729,7 @@ def transform(self, func, *args, **kwargs):
         1  1.000000   2.718282
         2  1.414214   7.389056
         """
-        result = self.agg(func, *args, **kwargs)
-        if is_scalar(result) or len(result) != len(self):
-            raise ValueError("transforms cannot produce aggregated results")
-
-        return result
+        raise NotImplementedError
path: pandas/core/generic.py
text: @user1:
can we remove this? or is the doc-string used?

@user1:
maybe just move the doc-string to shared_docs?

diff_hunk: @@ -0,0 +1,184 @@
+import operator
+import re
+
+import numpy as np
+import pytest
+
+from pandas import DataFrame, MultiIndex
+import pandas._testing as tm
+from pandas.core.base import SpecificationError
+from pandas.core.groupby.base import transformation_kernels
+from pandas.tests.frame.common import zip_frames
+
+
+def test_transform(axis, float_frame):
+    # GH 35964
+    other_axis = 1 if axis in {0, "index"} else 0
+
+    with np.errstate(all="ignore"):
+        f_abs = np.abs(float_frame)
+        f_sqrt = np.sqrt(float_frame)
+
+        # ufunc
+        result = float_frame.transform(np.sqrt, axis=axis)
+        expected = f_sqrt
+        tm.assert_frame_equal(result, expected)
+
+        # list-like
+        expected = f_sqrt.copy()
+        if axis in {0, "index"}:
+            expected.columns = MultiIndex.from_product([float_frame.columns, ["sqrt"]])
+        else:
+            expected.index = MultiIndex.from_product([float_frame.index, ["sqrt"]])
+        result = float_frame.transform([np.sqrt], axis=axis)
+        tm.assert_frame_equal(result, expected)
+
+        # multiple items in list
+        # these are in the order as if we are applying both
+        # functions per series and then concatting
+        expected = zip_frames([f_abs, f_sqrt], axis=other_axis)
+        if axis in {0, "index"}:
+            expected.columns = MultiIndex.from_product(
+                [float_frame.columns, ["absolute", "sqrt"]]
+            )
+        else:
+            expected.index = MultiIndex.from_product(
+                [float_frame.index, ["absolute", "sqrt"]]
+            )
+        result = float_frame.transform([np.abs, "sqrt"], axis=axis)
+        tm.assert_frame_equal(result, expected)
+
+
+@pytest.mark.parametrize("use_apply", [True, False])
+def test_transform_udf(axis, float_frame, use_apply):
+    # GH 35964
+    # transform uses UDF either via apply or passing the entire DataFrame
+    def func(x):
+        # transform is using apply iff x is not a DataFrame
+        if use_apply == isinstance(x, DataFrame):
+            # Force transform to fallback
+            raise ValueError
+        return x + 1
+
+    result = float_frame.transform(func, axis=axis)
+    expected = float_frame + 1
+    tm.assert_frame_equal(result, expected)
+
+
+@pytest.mark.parametrize("method", ["abs", "shift", "pct_change", "cumsum", "rank"])
+def test_transform_method_name(method):
+    # GH 19760
+    df = DataFrame({"A": [-1, 2]})
+    result = df.transform(method)
+    expected = operator.methodcaller(method)(df)
+    tm.assert_frame_equal(result, expected)
+
+
+def test_transform_and_agg_err(axis, float_frame):
+    # GH 35964
+    # cannot both transform and agg
+    msg = "Function did not transform"
+    with pytest.raises(ValueError, match=msg):
+        float_frame.transform(["max", "min"], axis=axis)
+
+    msg = "Function did not transform"
+    with pytest.raises(ValueError, match=msg):
+        float_frame.transform(["max", "sqrt"], axis=axis)
+
+
+def test_agg_dict_nested_renaming_depr():
+    df = DataFrame({"A": range(5), "B": 5})
+
+    # nested renaming
+    msg = r"nested renamer is not supported"
+    with pytest.raises(SpecificationError, match=msg):
+        df.transform({"A": {"foo": "min"}, "B": {"bar": "max"}})
+
+
+def test_transform_reducer_raises(all_reductions):
+    # GH 35964
+    op = all_reductions
+    df = DataFrame({"A": [1, 2, 3]})
+    msg = "Function did not transform"
+    with pytest.raises(ValueError, match=msg):
+        df.transform(op)
+    with pytest.raises(ValueError, match=msg):
+        df.transform([op])
+    with pytest.raises(ValueError, match=msg):
+        df.transform({"A": op})
+    with pytest.raises(ValueError, match=msg):
+        df.transform({"A": [op]})
+
+
+# mypy doesn't allow adding lists of different types
+# https://github.com/python/mypy/issues/5492
+@pytest.mark.parametrize("op", [*transformation_kernels, lambda x: x + 1])
+def test_transform_bad_dtype(op):
+    # GH 35964
+    df = DataFrame({"A": 3 * [object]})  # DataFrame that will fail on most transforms
+    if op in ("backfill", "shift", "pad", "bfill", "ffill"):
+        pytest.xfail("Transform function works on any datatype")
+    msg = "Transform function failed"
+    with pytest.raises(ValueError, match=msg):
+        df.transform(op)
+    with pytest.raises(ValueError, match=msg):
+        df.transform([op])
+    with pytest.raises(ValueError, match=msg):
+        df.transform({"A": op})
+    with pytest.raises(ValueError, match=msg):
+        df.transform({"A": [op]})
+
+
+@pytest.mark.parametrize("op", transformation_kernels)
+def test_transform_multi_dtypes(op):
+    # GH 35964
+    df = DataFrame({"A": ["a", "b", "c"], "B": [1, 2, 3]})
+
+    # Determine which columns op will work on
+    columns = []
+    for column in df:
path: pandas/tests/frame/apply/test_frame_transform.py
text: @user1:
hmm can we explicit list the exluded ones rather than this?

diff_hunk: @@ -384,3 +385,102 @@ def validate_func_kwargs(
     if not columns:
         raise TypeError(no_arg_message)
     return columns, func
+
+
+def transform(
+    obj: FrameOrSeries,
+    func: Union[str, List, Dict, Callable],
+    axis: int,
+    *args,
+    **kwargs,
+) -> FrameOrSeries:
+    """
+    Transform a DataFrame or Series
+
+    Parameters
+    ----------
+    obj : DataFrame or Series
+        Object to compute the transform on.
+    func : string, function, list, or dictionary
+        Function(s) to compute the transform with.
+    axis : {0 or 'index', 1 or 'columns'}
+        Axis along which the function is applied:
+
+        * 0 or 'index': apply function to each column.
+        * 1 or 'columns': apply function to each row.
+
+    Returns
+    -------
+    DataFrame or Series
+        Result of applying ``func`` along the given axis of the
+        Series or DataFrame.
+
+    Raises
+    ------
+    ValueError
+        If the transform function fails or does not transform.
+    """
+    is_series = obj.ndim == 1
+
+    if obj._get_axis_number(axis) == 1:
+        assert not is_series
+        return transform(obj.T, func, 0, *args, **kwargs).T
+
+    if isinstance(func, list):
+        if is_series:
+            func = {com.get_callable_name(v) or v: v for v in func}
+        else:
+            func = {col: func for col in obj}
+
+    if isinstance(func, dict):
+        if not is_series:
+            cols = sorted(set(func.keys()) - set(obj.columns))
+            if len(cols) > 0:
+                raise SpecificationError(f"Column(s) {cols} do not exist")
+
+        if any(isinstance(v, dict) for v in func.values()):
+            # GH 15931 - deprecation of renaming keys
+            raise SpecificationError("nested renamer is not supported")
+
+        results = {}
+        for name, how in func.items():
+            colg = obj._gotitem(name, ndim=1)
+            try:
+                results[name] = transform(colg, how, 0, *args, **kwargs)
+            except Exception as e:
+                if str(e) == "Function did not transform":
+                    raise e
+
+        # combine results
+        if len(results) == 0:
+            raise ValueError("Transform function failed")
+        from pandas.core.reshape.concat import concat
path: pandas/core/aggregation.py
text: @user1:
put this at the top of the function