<commit_before># Copyright 2015 Google Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
# in compliance with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the License
# is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
# or implied. See the License for the specific language governing permissions and limitations under
# the License.

"""Google Cloud Platform library - BigQuery IPython Functionality."""
from builtins import str
import google
import google.datalab.utils as utils

# TODO(rajivpb): These contrib imports are a stop-gap for
# https://github.com/googledatalab/pydatalab/issues/593
from google.datalab.contrib.pipeline._pipeline import Pipeline
from google.datalab.contrib.pipeline.airflow._airflow import Airflow

import argparse
import jsonschema


def _create_pipeline_subparser(parser):
  pipeline_parser = parser.subcommand(
    'pipeline',
    formatter_class=argparse.RawTextHelpFormatter,
    help="""
Creates a GCS/BigQuery ETL pipeline. The cell-body is specified as follows:
  input:
    table | path: <BQ table name or GCS path>; both if path->table load is required
    schema: <syntax is same as that in '%%bq execute'>
    format: {csv (default) | json}
    csv: (if 'csv' is the 'format')
      delimiter: the field delimiter to use. Defaults to ','.
      skip: Number of rows at the top of a CSV file to skip. Defaults to 0.
      strict: If False (default), does not accept rows with missing trailing optional columns.
      quote: The value used to quote data sections. Defaults to '"'.
    mode: {append (default) | overwrite}; required if loading from path to table.
  transformation:
    query: <name of BQ query defined via "%%bq query --name ...">
  output:
    table | path: <BQ table name or GCS path>; both if table->path extract is required
    format: {csv (default) | json}
    csv: (if 'csv' is the 'format')
      delimiter: the field delimiter to use. Defaults to ','
      header: {True (default) | False}; Whether to include an initial header line.
      compress: {True | False (default) }; Whether to compress the data on export.
  schedule: (optional; defaults below will apply when omitted)
    start: <formatted as '%Y-%m-%dT%H:%M:%S'; default is 'now'>
    end:  <formatted as '%Y-%m-%dT%H:%M:%S'; default is 'forever'>
    interval: {@once (default) | @hourly | @daily | @weekly | @ monthly | @yearly | <cron ex>}
  parameters: <syntax is same as that in '%%bq execute'>')
""")

  pipeline_parser.add_argument('-n', '--name', type=str, help='BigQuery pipeline name',
                               required=True)
  pipeline_parser.add_argument('-d', '--gcs_dag_bucket', type=str,
                               help='The Google Cloud Storage bucket for the Airflow dags.')
  pipeline_parser.add_argument('-f', '--gcs_dag_file_path', type=str,
                               help='The file path suffix for the Airflow dags.')
  return pipeline_parser


def _pipeline_cell(args, cell_body):
    """Implements the pipeline subcommand in the %%bq magic.
    Args:
      args: the arguments following '%%bq pipeline'.
      cell_body: Cell contents.
    """
    name = args.get('name')
    if name is None:
        raise Exception('Pipeline name was not specified.')

    bq_pipeline_config = utils.commands.parse_config(
        cell_body, utils.commands.notebook_environment())
    pipeline_spec = _get_pipeline_spec_from_config(bq_pipeline_config)

    pipeline = Pipeline(name, pipeline_spec)
    utils.commands.notebook_environment()[name] = pipeline

    airflow_spec = pipeline.get_airflow_spec()<commit_msg>Fix reference documentation for schedule section; add 'catchup' description (#657)<commit_after># Copyright 2015 Google Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
# in compliance with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the License
# is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
# or implied. See the License for the specific language governing permissions and limitations under
# the License.

"""Google Cloud Platform library - BigQuery IPython Functionality."""
from builtins import str
import google
import google.datalab.utils as utils

# TODO(rajivpb): These contrib imports are a stop-gap for
# https://github.com/googledatalab/pydatalab/issues/593
from google.datalab.contrib.pipeline._pipeline import Pipeline
from google.datalab.contrib.pipeline.airflow._airflow import Airflow

import argparse
import jsonschema


def _create_pipeline_subparser(parser):
  pipeline_parser = parser.subcommand(
    'pipeline',
    formatter_class=argparse.RawTextHelpFormatter,
    help="""
Creates a GCS/BigQuery ETL pipeline. The cell-body is specified as follows:
  input:
    table | path: <BQ table name or GCS path; both if path->table load is also required>
    schema: <For syntax, refer '%%bq execute'>
    format: {csv (default) | json}
    csv: <This section is relevant only when 'format' is 'csv'>
      delimiter: <The field delimiter to use; default is ','>
      skip: <Number of rows at the top of a CSV file to skip; default is 0>
      strict: <{True | False (default)}; whether to accept rows with missing trailing (or optional) columns>
      quote: <Value used to quote data sections; default is '"'>
    mode: <{append (default) | overwrite}; required if path->table load>
  transformation: <optional; when absent, a direct conversion is done from input (path|table) to output (table|path)>
    query: <name of BQ query defined via "%%bq query --name ...">
  output:
    table | path: <BQ table name or GCS path; both if table->path extract is required>
    format: <{csv (default) | json}>
    csv: <This section is relevant only when 'format' is 'csv'>
      delimiter: <the field delimiter to use. Defaults to ','>
      header: <{True (default) | False}; Whether to include an initial header line>
      compress: <{True | False (default) }; Whether to compress the data on export>
  schedule:
    start: <formatted as '%Y-%m-%dT%H:%M:%S'; default is 'now'>
    end:  <formatted as '%Y-%m-%dT%H:%M:%S'; default is 'forever'>
    interval: <{@once (default) | @hourly | @daily | @weekly | @ monthly | @yearly | <cron ex>}>
    catchup: <{True | False (default)}; Whether to have pipelines execute in the past>
  parameters: <For syntax, refer '%%bq execute'>
""")  # noqa

  pipeline_parser.add_argument('-n', '--name', type=str, help='BigQuery pipeline name',
                               required=True)
  pipeline_parser.add_argument('-d', '--gcs_dag_bucket', type=str,
                               help='The Google Cloud Storage bucket for the Airflow dags.')
  pipeline_parser.add_argument('-f', '--gcs_dag_file_path', type=str,
                               help='The file path suffix for the Airflow dags.')
  return pipeline_parser


def _pipeline_cell(args, cell_body):
    """Implements the pipeline subcommand in the %%bq magic.
    Args:
      args: the arguments following '%%bq pipeline'.
      cell_body: Cell contents.
    """
    name = args.get('name')
    if name is None:
        raise Exception('Pipeline name was not specified.')

    bq_pipeline_config = utils.commands.parse_config(
        cell_body, utils.commands.notebook_environment())
    pipeline_spec = _get_pipeline_spec_from_config(bq_pipeline_config)

    pipeline = Pipeline(name, pipeline_spec)
    utils.commands.notebook_environment()[name] = pipeline

    airflow_spec = pipeline.get_airflow_spec()
{'fraction_non_alphanumeric': 0.10444752425349628, 'fraction_numerical': 0.003401789089076477, 'mean_word_length': 3.633975481611208, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 38, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5558794', 'n_tokens_mistral': 2296, 'n_tokens_neox': 2277, 'n_words': 866}