**Description**

Similarly to `@dag` and `@task` decorators, we can add a decorator to easily generate `TaskGroup` instances. 

**Use case / motivation**

This would be used for reusable pieces of DAGs. An example may be to hyperparameter tune an ML pipeline. You may want to generate several parallel subpipelines with different settings/parameters to be executed in parallel. At the end pull the result and decide the best parametrization.  

```py
@task
collect_dataset(...)
  pass

@task
def train_model(...)
   pass

@task_group
def movielens_model_pipeline(learning_rate: int, feature_list: list, dataset: XComArg):  
  dataset = filter_dataset(dataset, feature_list)
  train_model(dataset, learning_rate)

@dag
def movielens_hpt(dataset_path: str, feature_list:list=['year', 'director']):
  dataset = load_dataset(dataset_path)
  for i in range(0.1, 0.9, step=0.1):
    movielens_model_pipeline(i, feature_list, dataset)
  
  decide_best_model(...)
```




**Related Issues**

https://github.com/apache/airflow/issues?q=is%3Aopen+is%3Aissue+label%3AAIP-31


airflow/utils/task_group.py
raise ValueError("group_id must be str")
            if not group_id:
                raise ValueError("group_id must not be empty")
            dag = dag or DagContext.get_current_dag()
            if not parent_group and not dag:
                raise AirflowException("TaskGroup can only be used inside a dag")
            self._parent_group = parent_group or TaskGroupContext.get_current_task_group(dag)
            if not self._parent_group:
                raise AirflowException("TaskGroup must have a parent_group except for the root TaskGroup")
            self.used_group_ids = self._parent_group.used_group_ids
        self._group_id = group_id
        if self.group_id in self.used_group_ids:
            raise DuplicateTaskIdFound(f"group_id '{self.group_id}' has already been added to the DAG")
        self.used_group_ids.add(self.group_id)
        self.used_group_ids.add(self.downstream_join_id)
        self.used_group_ids.add(self.upstream_join_id)
        self.children: Dict[str, Union["BaseOperator", "TaskGroup"]] = {}
        if self._parent_group:
            self._parent_group.add(self)
        self.tooltip = tooltip
        self.ui_color = ui_color
        self.ui_fgcolor = ui_fgcolor
        
        
        self.upstream_group_ids: Set[Optional[str]] = set()
        self.downstream_group_ids: Set[Optional[str]] = set()
        self.upstream_task_ids: Set[Optional[str]] = set()
        self.downstream_task_ids: Set[Optional[str]] = set()
    @classmethod
    def create_root(cls, dag: "DAG") -> "TaskGroup":