    def map(self, input_schema, output_schema, fn, config=None) -> 'Pipeline':
        """
        One to one map of function on inputs.

        Args:
            input_schema (tuple): The input column/s of fn.
            output_schema (tuple): The output column/s of fn.
            fn (Operation | lambda | callable): The action to perform on the input_schema.
            config (dict, optional): Config for the map. Defaults to None.

        Returns:
            Pipeline: Pipeline with action added.

        Examples:
            >>> from towhee import pipe
            >>> p = pipe.input('a').map('a', 'b', lambda x: x+1).output('a', 'b')
            >>> p(1).get()
            [1, 2]
        """
        output_schema = self._check_schema(output_schema)
        input_schema = self._check_schema(input_schema)

        uid = uuid.uuid4().hex
        fn_action = self._to_action(fn)
        dag_dict = deepcopy(self._dag)
        dag_dict[uid] = {
            'inputs': input_schema,
            'outputs': output_schema,
            'op_info': fn_action.serialize(),
            'iter_info': {
                'type': MapConst.name,
                'param': None
            },
            'config': config,
            'next_nodes': [],
        }
        dag_dict[self._clo_node]['next_nodes'].append(uid)
        return Pipeline(dag_dict, uid)
One to one map of function on inputs.

Args:
input_schema (tuple): The input column/s of fn.
output_schema (tuple): The output column/s of fn.
fn (Operation | lambda | callable): The action to perform on the input_schema.
config (dict, optional): Config for the map. Defaults to None.

Returns:
Pipeline: Pipeline with action added.

Examples:
>>> from towhee import pipe
>>> p = pipe.input('a').map('a', 'b', lambda x: x+1).output('a', 'b')
>>> p(1).get()
[1, 2]
towhee/runtime/pipeline.py