def insert_all(self, project_id, dataset_id, table_id,
                   rows, ignore_unknown_values=False,
                   skip_invalid_rows=False, fail_on_error=False):
        """
        
        """

        dataset_project_id = project_id if project_id else self.project_id

        body = {
            "rows": rows,
            "ignoreUnknownValues": ignore_unknown_values,
            "kind": "bigquery#tableDataInsertAllRequest",
            "skipInvalidRows": skip_invalid_rows,
        }

        try:
            self.log.info(
                'Inserting %s row(s) into Table %s:%s.%s',
                len(rows), dataset_project_id, dataset_id, table_id
            )

            resp = self.service.tabledata().insertAll(
                projectId=dataset_project_id, datasetId=dataset_id,
                tableId=table_id, body=body
            ).execute(num_retries=self.num_retries)

            if 'insertErrors' not in resp:
                self.log.info(
                    'All row(s) inserted successfully: %s:%s.%s',
                    dataset_project_id, dataset_id, table_id
                )
            else:
                error_msg = '{} insert error(s) occurred: {}:{}.{}. Details: {}'.format(
                    len(resp['insertErrors']),
                    dataset_project_id, dataset_id, table_id, resp['insertErrors'])
                if fail_on_error:
                    raise AirflowException(
                        'BigQuery job failed. Error was: {}'.format(error_msg)
                    )
                self.log.info(error_msg)
        except HttpError as err:
            raise AirflowException(
                'BigQuery job failed. Error was: {}'.format(err.content)
            )