1528

pending completion

Build # 1528

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/googlegenomics/gcp-variant-transforms/commit/<a class=hub.com/googlegenomics/gcp-variant-transforms/commit/<a class="double-link" href="https://git"><a class=hub.com/googlegenomics/gcp-variant-transforms/commit/93b8cd3832725f9a4856aac90beb6f2ee75923ca">93b8cd383<a href="https://github.com/googlegenomics/gcp-variant-transforms/commit/93b8cd3832725f9a4856aac90beb6f2ee75923ca">&lt;a href=&quot;https://github.com/googlegenomics/gcp-variant-transforms/commit/93b8cd3832725f9a4856aac90beb6f2ee75923ca&quot;&gt;&amp;quot;&amp;gt;Revert &amp;quot;Scalable vep annotation in VT&amp;quot; (#432)

* Revert &amp;quot;Extract common logic out of merge_header_definitions.py (#431)&amp;quot;

This reverts commit &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/googlegenomics/gcp-variant-transforms/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/googlegenomics/gcp-variant-transforms/commit/49131f3b910a715e82831f59b71c3db0baae2677&quot;&gt;49131f3b9&lt;/a&gt;&quot;&gt;49131f3b9</a><a href="https://github.com/googlegenomics/gcp-variant-transforms/commit/93b8cd3832725f9a4856aac90beb6f2ee75923ca">.

* Revert &quot;Scalable vep annotation in VT (#430)&quot;

This reverts commit </a><a class="double-link" href="https://github.com/googlegenomics/gcp-variant-transforms/commit/<a class="double-link" href="https://github.com/googlegenomics/gcp-variant-transforms/commit/933b51611a4fefaac53b336b99076bcc2701ce6a">933b51611">933b51611.

Run Details

52 of 52 new or added lines in 7 files covered. (100.0%)

6742 of 7549 relevant lines covered (89.31%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

35.16

/gcp_variant_transforms/vcf_to_bq.py

# Copyright 2017 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.

r"""Pipeline for loading VCF files to BigQuery.

Run locally:
python -m gcp_variant_transforms.vcf_to_bq \
  --input_pattern <path to VCF file(s)> \
  --output_table projectname:bigquerydataset.tablename

Run on Dataflow:
python -m gcp_variant_transforms.vcf_to_bq \
  --input_pattern gs://bucket/vcfs/vcffile.vcf \
  --output_table projectname:bigquerydataset.tablename \
  --project projectname \
  --staging_location gs://bucket/staging \
  --temp_location gs://bucket/temp \
  --job_name vcf-to-bq \
  --setup_file ./setup.py \
  --runner DataflowRunner
"""

from __future__ import absolute_import

import argparse  # pylint: disable=unused-import
import logging
import sys
import tempfile
from typing import List, Optional  # pylint: disable=unused-import

import apache_beam as beam
from apache_beam import pvalue  # pylint: disable=unused-import
from apache_beam.io import filesystems
from apache_beam.options import pipeline_options

from gcp_variant_transforms import pipeline_common
from gcp_variant_transforms.beam_io import vcfio
from gcp_variant_transforms.libs import metrics_util
from gcp_variant_transforms.libs import processed_variant
from gcp_variant_transforms.libs import vcf_header_parser
from gcp_variant_transforms.libs import variant_partition
from gcp_variant_transforms.libs.annotation.vep import vep_runner
from gcp_variant_transforms.libs.variant_merge import merge_with_non_variants_strategy
from gcp_variant_transforms.libs.variant_merge import move_to_calls_strategy
from gcp_variant_transforms.libs.variant_merge import variant_merge_strategy  # pylint: disable=unused-import
from gcp_variant_transforms.options import variant_transform_options
from gcp_variant_transforms.transforms import filter_variants
from gcp_variant_transforms.transforms import infer_headers
from gcp_variant_transforms.transforms import merge_headers
from gcp_variant_transforms.transforms import merge_variants
from gcp_variant_transforms.transforms import partition_variants
from gcp_variant_transforms.transforms import variant_to_avro
from gcp_variant_transforms.transforms import variant_to_bigquery


_COMMAND_LINE_OPTIONS = [
    variant_transform_options.VcfReadOptions,
    variant_transform_options.AvroWriteOptions,
    variant_transform_options.BigQueryWriteOptions,
    variant_transform_options.AnnotationOptions,
    variant_transform_options.FilterOptions,
    variant_transform_options.MergeOptions,
    variant_transform_options.PartitionOptions,
]

_MERGE_HEADERS_FILE_NAME = 'merged_headers.vcf'
_MERGE_HEADERS_JOB_NAME = 'merge-vcf-headers'


def _read_variants(pipeline, known_args):
  # type: (beam.Pipeline, argparse.Namespace) -> pvalue.PCollection
  """Helper method for returning a PCollection of Variants from VCFs."""
  representative_header_lines = None
  if known_args.representative_header_file:
    representative_header_lines = vcf_header_parser.get_metadata_header_lines(
        known_args.representative_header_file)

  if known_args.optimize_for_large_inputs:
    variants = (pipeline
                | 'InputFilePattern' >> beam.Create([known_args.input_pattern])
                | 'ReadAllFromVcf' >> vcfio.ReadAllFromVcf(
                    representative_header_lines=representative_header_lines,
                    allow_malformed_records=(
                        known_args.allow_malformed_records)))
  else:
    variants = pipeline | 'ReadFromVcf' >> vcfio.ReadFromVcf(
        known_args.input_pattern,
        representative_header_lines=representative_header_lines,
        allow_malformed_records=known_args.allow_malformed_records,
        vcf_parser_type=vcfio.VcfParserType[known_args.vcf_parser])
  return variants


def _get_variant_merge_strategy(known_args  # type: argparse.Namespace
                               ):
  # type: (...) -> Optional[variant_merge_strategy.VariantMergeStrategy]
  merge_options = variant_transform_options.MergeOptions
  if (not known_args.variant_merge_strategy or
      known_args.variant_merge_strategy == merge_options.NONE):
    return None
  elif known_args.variant_merge_strategy == merge_options.MOVE_TO_CALLS:
    return move_to_calls_strategy.MoveToCallsStrategy(
        known_args.info_keys_to_move_to_calls_regex,
        known_args.copy_quality_to_calls,
        known_args.copy_filter_to_calls)
  elif (known_args.variant_merge_strategy ==
        merge_options.MERGE_WITH_NON_VARIANTS):
    return merge_with_non_variants_strategy.MergeWithNonVariantsStrategy(
        known_args.info_keys_to_move_to_calls_regex,
        known_args.copy_quality_to_calls,
        known_args.copy_filter_to_calls)
  else:
    raise ValueError('Merge strategy is not supported.')


def _add_inferred_headers(pipeline,  # type: beam.Pipeline
                          known_args,  # type: argparse.Namespace
                          merged_header  # type: pvalue.PCollection
                         ):
  # type: (...) -> pvalue.PCollection
  annotation_fields_to_infer = (known_args.annotation_fields if
                                known_args.infer_annotation_types else [])
  inferred_headers = (
      _read_variants(pipeline, known_args)
      | 'FilterVariants' >> filter_variants.FilterVariants(
          reference_names=known_args.reference_names)
      | 'InferHeaderFields' >> infer_headers.InferHeaderFields(
          pvalue.AsSingleton(merged_header),
          known_args.allow_incompatible_records,
          known_args.infer_headers,
          annotation_fields_to_infer))
  merged_header = (
      (inferred_headers, merged_header)
      | beam.Flatten()
      | 'MergeHeadersFromVcfAndVariants' >> merge_headers.MergeHeaders(
          known_args.split_alternate_allele_info_fields,
          known_args.allow_incompatible_records))
  return merged_header


def _merge_headers(known_args, pipeline_args, pipeline_mode):
  # type: (argparse.Namespace, List[str], int) -> None
  """Merges VCF headers using beam based on pipeline_mode."""
  if known_args.representative_header_file:
    return

  options = pipeline_options.PipelineOptions(pipeline_args)

  # Always run pipeline locally if data is small.
  if (pipeline_mode == pipeline_common.PipelineModes.SMALL and
      not known_args.infer_headers and not known_args.infer_annotation_types):
    options.view_as(pipeline_options.StandardOptions).runner = 'DirectRunner'

  google_cloud_options = options.view_as(pipeline_options.GoogleCloudOptions)
  merge_headers_job_name = pipeline_common.generate_unique_name(
      _MERGE_HEADERS_JOB_NAME)
  if google_cloud_options.job_name:
    google_cloud_options.job_name += '-' + merge_headers_job_name
  else:
    google_cloud_options.job_name = merge_headers_job_name

  temp_directory = google_cloud_options.temp_location or tempfile.mkdtemp()
  temp_merged_headers_file_name = '-'.join([google_cloud_options.job_name,
                                            _MERGE_HEADERS_FILE_NAME])
  temp_merged_headers_file_path = filesystems.FileSystems.join(
      temp_directory, temp_merged_headers_file_name)

  with beam.Pipeline(options=options) as p:
    headers = pipeline_common.read_headers(p, pipeline_mode, known_args)
    merged_header = pipeline_common.get_merged_headers(
        headers,
        known_args.split_alternate_allele_info_fields,
        known_args.allow_incompatible_records)
    if known_args.infer_headers or known_args.infer_annotation_types:
      merged_header = _add_inferred_headers(p, known_args, merged_header)
    pipeline_common.write_headers(merged_header, temp_merged_headers_file_path)
    known_args.representative_header_file = temp_merged_headers_file_path


def run(argv=None):
  # type: (List[str]) -> None
  """Runs VCF to BigQuery pipeline."""
  logging.info('Command: %s', ' '.join(argv or sys.argv))
  known_args, pipeline_args = pipeline_common.parse_args(argv,
                                                         _COMMAND_LINE_OPTIONS)
  # Note VepRunner creates new input files, so it should be run before any
  # other access to known_args.input_pattern.
  if known_args.run_annotation_pipeline:
    runner = vep_runner.create_runner_and_update_args(known_args, pipeline_args)
    runner.run_on_all_files()
    runner.wait_until_done()
    logging.info('Using VEP processed files: %s', known_args.input_pattern)

  variant_merger = _get_variant_merge_strategy(known_args)
  pipeline_mode = pipeline_common.get_pipeline_mode(
      known_args.input_pattern, known_args.optimize_for_large_inputs)

  # Starts a pipeline to merge VCF headers in beam if the total files that
  # match the input pattern exceeds _SMALL_DATA_THRESHOLD
  _merge_headers(known_args, pipeline_args, pipeline_mode)

  # Retrieve merged headers prior to launching the pipeline. This is needed
  # since the BigQuery schema cannot yet be dynamically created based on input.
  # See https://issues.apache.org/jira/browse/BEAM-2801.
  header_fields = vcf_header_parser.get_vcf_headers(
      known_args.representative_header_file)
  counter_factory = metrics_util.CounterFactory()
  processed_variant_factory = processed_variant.ProcessedVariantFactory(
      header_fields,
      known_args.split_alternate_allele_info_fields,
      known_args.allow_malformed_records,
      known_args.annotation_fields,
      known_args.use_allele_num,
      known_args.minimal_vep_alt_matching,
      known_args.infer_annotation_types,
      counter_factory)

  partitioner = None
  if ((known_args.optimize_for_large_inputs and variant_merger) or
      known_args.partition_config_path):
    partitioner = variant_partition.VariantPartition(
        known_args.partition_config_path)

  beam_pipeline_options = pipeline_options.PipelineOptions(pipeline_args)
  pipeline = beam.Pipeline(options=beam_pipeline_options)
  variants = _read_variants(pipeline, known_args)
  variants |= 'FilterVariants' >> filter_variants.FilterVariants(
      reference_names=known_args.reference_names)
  if partitioner:
    num_partitions = partitioner.get_num_partitions()
    partitioned_variants = variants | 'PartitionVariants' >> beam.Partition(
        partition_variants.PartitionVariants(partitioner), num_partitions)
    variants = []
    for i in range(num_partitions):
      if partitioner.should_keep_partition(i):
        variants.append(partitioned_variants[i])
      else:
        num_partitions -= 1
  else:
    # By default we don't partition the data, so we have only 1 partition.
    num_partitions = 1
    variants = [variants]

  for i in range(num_partitions):
    if variant_merger:
      variants[i] |= ('MergeVariants' + str(i) >>
                      merge_variants.MergeVariants(variant_merger))
    variants[i] |= (
        'ProcessVariants' + str(i) >>
        beam.Map(processed_variant_factory.create_processed_variant).\
            with_output_types(processed_variant.ProcessedVariant))
  if partitioner and partitioner.should_flatten():
    variants = [variants | 'FlattenPartitions' >> beam.Flatten()]
    num_partitions = 1

  if known_args.output_table:
    for i in range(num_partitions):
      table_suffix = ''
      if partitioner and partitioner.get_partition_name(i):
        table_suffix = '_' + partitioner.get_partition_name(i)
      table_name = known_args.output_table + table_suffix
      _ = (variants[i] | 'VariantToBigQuery' + table_suffix >>
           variant_to_bigquery.VariantToBigQuery(
               table_name,
               header_fields,
               variant_merger,
               processed_variant_factory,
               append=known_args.append,
               update_schema_on_append=known_args.update_schema_on_append,
               allow_incompatible_records=known_args.allow_incompatible_records,
               omit_empty_sample_calls=known_args.omit_empty_sample_calls,
               num_bigquery_write_shards=known_args.num_bigquery_write_shards,
               null_numeric_value_replacement=(
                   known_args.null_numeric_value_replacement)))

  if known_args.output_avro_path:
    # TODO(bashir2): Add an integration test that outputs to Avro files and
    # also imports to BigQuery. Then import those Avro outputs using the bq
    # tool and verify that the two tables are identical.
    _ = (
        variants | 'FlattenToOnePCollection' >> beam.Flatten()
        | 'VariantToAvro' >>
        variant_to_avro.VariantToAvroFiles(
            known_args.output_avro_path,
            header_fields,
            processed_variant_factory,
            variant_merger=variant_merger,
            allow_incompatible_records=known_args.allow_incompatible_records,
            omit_empty_sample_calls=known_args.omit_empty_sample_calls,
            null_numeric_value_replacement=(
                known_args.null_numeric_value_replacement))
    )

  result = pipeline.run()
  result.wait_until_finish()

  metrics_util.log_all_counters(result)


if __name__ == '__main__':
  logging.getLogger().setLevel(logging.INFO)
  run()

1	# Copyright 2017 Google Inc. All Rights Reserved.
2	#
3	# Licensed under the Apache License, Version 2.0 (the "License");
4	# you may not use this file except in compliance with the License.
5	# You may obtain a copy of the License at
6	#
7	# http://www.apache.org/licenses/LICENSE-2.0
8	#
9	# Unless required by applicable law or agreed to in writing, software
10	# distributed under the License is distributed on an "AS IS" BASIS,
11	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	# See the License for the specific language governing permissions and
13	# limitations under the License.
14
15	r"""Pipeline for loading VCF files to BigQuery.	1✔
16
17	Run locally:
18	python -m gcp_variant_transforms.vcf_to_bq \
19	--input_pattern <path to VCF file(s)> \
20	--output_table projectname:bigquerydataset.tablename
21
22	Run on Dataflow:
23	python -m gcp_variant_transforms.vcf_to_bq \
24	--input_pattern gs://bucket/vcfs/vcffile.vcf \
25	--output_table projectname:bigquerydataset.tablename \
26	--project projectname \
27	--staging_location gs://bucket/staging \
28	--temp_location gs://bucket/temp \
29	--job_name vcf-to-bq \
30	--setup_file ./setup.py \
31	--runner DataflowRunner
32	"""
33
34	from __future__ import absolute_import	1✔
35
36	import argparse # pylint: disable=unused-import	1✔
37	import logging	1✔
38	import sys	1✔
39	import tempfile	1✔
40	from typing import List, Optional # pylint: disable=unused-import	1✔
41
42	import apache_beam as beam	1✔
43	from apache_beam import pvalue # pylint: disable=unused-import	1✔
44	from apache_beam.io import filesystems	1✔
45	from apache_beam.options import pipeline_options	1✔
46
47	from gcp_variant_transforms import pipeline_common	1✔
48	from gcp_variant_transforms.beam_io import vcfio	1✔
49	from gcp_variant_transforms.libs import metrics_util	1✔
50	from gcp_variant_transforms.libs import processed_variant	1✔
51	from gcp_variant_transforms.libs import vcf_header_parser	1✔
52	from gcp_variant_transforms.libs import variant_partition	1✔
53	from gcp_variant_transforms.libs.annotation.vep import vep_runner	1✔
54	from gcp_variant_transforms.libs.variant_merge import merge_with_non_variants_strategy	1✔
55	from gcp_variant_transforms.libs.variant_merge import move_to_calls_strategy	1✔
56	from gcp_variant_transforms.libs.variant_merge import variant_merge_strategy # pylint: disable=unused-import	1✔
57	from gcp_variant_transforms.options import variant_transform_options	1✔
58	from gcp_variant_transforms.transforms import filter_variants	1✔
59	from gcp_variant_transforms.transforms import infer_headers	1✔
60	from gcp_variant_transforms.transforms import merge_headers	1✔
61	from gcp_variant_transforms.transforms import merge_variants	1✔
62	from gcp_variant_transforms.transforms import partition_variants	1✔
63	from gcp_variant_transforms.transforms import variant_to_avro	1✔
64	from gcp_variant_transforms.transforms import variant_to_bigquery	1✔
65
66
67	_COMMAND_LINE_OPTIONS = [	1✔
68	variant_transform_options.VcfReadOptions,
69	variant_transform_options.AvroWriteOptions,
70	variant_transform_options.BigQueryWriteOptions,
71	variant_transform_options.AnnotationOptions,
72	variant_transform_options.FilterOptions,
73	variant_transform_options.MergeOptions,
74	variant_transform_options.PartitionOptions,
75	]
76
77	_MERGE_HEADERS_FILE_NAME = 'merged_headers.vcf'	1✔
78	_MERGE_HEADERS_JOB_NAME = 'merge-vcf-headers'	1✔
79
80
81	def _read_variants(pipeline, known_args):	1✔
82	# type: (beam.Pipeline, argparse.Namespace) -> pvalue.PCollection
83	"""Helper method for returning a PCollection of Variants from VCFs."""
84	representative_header_lines = None	×
85	if known_args.representative_header_file:	×
86	representative_header_lines = vcf_header_parser.get_metadata_header_lines(	×
87	known_args.representative_header_file)
88
89	if known_args.optimize_for_large_inputs:	×
90	variants = (pipeline	×
91	\| 'InputFilePattern' >> beam.Create([known_args.input_pattern])
92	\| 'ReadAllFromVcf' >> vcfio.ReadAllFromVcf(
93	representative_header_lines=representative_header_lines,
94	allow_malformed_records=(
95	known_args.allow_malformed_records)))
96	else:
97	variants = pipeline \| 'ReadFromVcf' >> vcfio.ReadFromVcf(	×
98	known_args.input_pattern,
99	representative_header_lines=representative_header_lines,
100	allow_malformed_records=known_args.allow_malformed_records,
101	vcf_parser_type=vcfio.VcfParserType[known_args.vcf_parser])
102	return variants	×
103
104
105	def _get_variant_merge_strategy(known_args # type: argparse.Namespace	1✔
106	):
107	# type: (...) -> Optional[variant_merge_strategy.VariantMergeStrategy]
108	merge_options = variant_transform_options.MergeOptions	1✔
109	if (not known_args.variant_merge_strategy or	1✔
110	known_args.variant_merge_strategy == merge_options.NONE):
111	return None	1✔
112	elif known_args.variant_merge_strategy == merge_options.MOVE_TO_CALLS:	1✔
113	return move_to_calls_strategy.MoveToCallsStrategy(	1✔
114	known_args.info_keys_to_move_to_calls_regex,
115	known_args.copy_quality_to_calls,
116	known_args.copy_filter_to_calls)
117	elif (known_args.variant_merge_strategy ==	1✔
118	merge_options.MERGE_WITH_NON_VARIANTS):
119	return merge_with_non_variants_strategy.MergeWithNonVariantsStrategy(	×
120	known_args.info_keys_to_move_to_calls_regex,
121	known_args.copy_quality_to_calls,
122	known_args.copy_filter_to_calls)
123	else:
124	raise ValueError('Merge strategy is not supported.')	1✔
125
126
127	def _add_inferred_headers(pipeline, # type: beam.Pipeline	1✔
128	known_args, # type: argparse.Namespace
129	merged_header # type: pvalue.PCollection
130	):
131	# type: (...) -> pvalue.PCollection
132	annotation_fields_to_infer = (known_args.annotation_fields if	×
133	known_args.infer_annotation_types else [])
134	inferred_headers = (	×
135	_read_variants(pipeline, known_args)
136	\| 'FilterVariants' >> filter_variants.FilterVariants(
137	reference_names=known_args.reference_names)
138	\| 'InferHeaderFields' >> infer_headers.InferHeaderFields(
139	pvalue.AsSingleton(merged_header),
140	known_args.allow_incompatible_records,
141	known_args.infer_headers,
142	annotation_fields_to_infer))
143	merged_header = (	×
144	(inferred_headers, merged_header)
145	\| beam.Flatten()
146	\| 'MergeHeadersFromVcfAndVariants' >> merge_headers.MergeHeaders(
147	known_args.split_alternate_allele_info_fields,
148	known_args.allow_incompatible_records))
149	return merged_header	×
150
151
152	def _merge_headers(known_args, pipeline_args, pipeline_mode):	1✔
153	# type: (argparse.Namespace, List[str], int) -> None
154	"""Merges VCF headers using beam based on pipeline_mode."""
155	if known_args.representative_header_file:	×
156	return	×
157
158	options = pipeline_options.PipelineOptions(pipeline_args)	×
159
160	# Always run pipeline locally if data is small.
161	if (pipeline_mode == pipeline_common.PipelineModes.SMALL and	×
162	not known_args.infer_headers and not known_args.infer_annotation_types):
163	options.view_as(pipeline_options.StandardOptions).runner = 'DirectRunner'	×
164
165	google_cloud_options = options.view_as(pipeline_options.GoogleCloudOptions)	×
166	merge_headers_job_name = pipeline_common.generate_unique_name(	×
167	_MERGE_HEADERS_JOB_NAME)
168	if google_cloud_options.job_name:	×
169	google_cloud_options.job_name += '-' + merge_headers_job_name	×
170	else:
171	google_cloud_options.job_name = merge_headers_job_name	×
172
173	temp_directory = google_cloud_options.temp_location or tempfile.mkdtemp()	×
174	temp_merged_headers_file_name = '-'.join([google_cloud_options.job_name,	×
175	_MERGE_HEADERS_FILE_NAME])
176	temp_merged_headers_file_path = filesystems.FileSystems.join(	×
177	temp_directory, temp_merged_headers_file_name)
178
179	with beam.Pipeline(options=options) as p:	×
180	headers = pipeline_common.read_headers(p, pipeline_mode, known_args)	×
181	merged_header = pipeline_common.get_merged_headers(	×
182	headers,
183	known_args.split_alternate_allele_info_fields,
184	known_args.allow_incompatible_records)
185	if known_args.infer_headers or known_args.infer_annotation_types:	×
186	merged_header = _add_inferred_headers(p, known_args, merged_header)	×
187	pipeline_common.write_headers(merged_header, temp_merged_headers_file_path)	×
188	known_args.representative_header_file = temp_merged_headers_file_path	×
189
190
191	def run(argv=None):	1✔
192	# type: (List[str]) -> None
193	"""Runs VCF to BigQuery pipeline."""
194	logging.info('Command: %s', ' '.join(argv or sys.argv))	×
195	known_args, pipeline_args = pipeline_common.parse_args(argv,	×
196	_COMMAND_LINE_OPTIONS)
197	# Note VepRunner creates new input files, so it should be run before any
198	# other access to known_args.input_pattern.
199	if known_args.run_annotation_pipeline:	×
200	runner = vep_runner.create_runner_and_update_args(known_args, pipeline_args)	×
201	runner.run_on_all_files()	×
202	runner.wait_until_done()	×
203	logging.info('Using VEP processed files: %s', known_args.input_pattern)	×
204
205	variant_merger = _get_variant_merge_strategy(known_args)	×
206	pipeline_mode = pipeline_common.get_pipeline_mode(	×
207	known_args.input_pattern, known_args.optimize_for_large_inputs)
208
209	# Starts a pipeline to merge VCF headers in beam if the total files that
210	# match the input pattern exceeds _SMALL_DATA_THRESHOLD
211	_merge_headers(known_args, pipeline_args, pipeline_mode)	×
212
213	# Retrieve merged headers prior to launching the pipeline. This is needed
214	# since the BigQuery schema cannot yet be dynamically created based on input.
215	# See https://issues.apache.org/jira/browse/BEAM-2801.
216	header_fields = vcf_header_parser.get_vcf_headers(	×
217	known_args.representative_header_file)
218	counter_factory = metrics_util.CounterFactory()	×
219	processed_variant_factory = processed_variant.ProcessedVariantFactory(	×
220	header_fields,
221	known_args.split_alternate_allele_info_fields,
222	known_args.allow_malformed_records,
223	known_args.annotation_fields,
224	known_args.use_allele_num,
225	known_args.minimal_vep_alt_matching,
226	known_args.infer_annotation_types,
227	counter_factory)
228
229	partitioner = None	×
230	if ((known_args.optimize_for_large_inputs and variant_merger) or	×
231	known_args.partition_config_path):
232	partitioner = variant_partition.VariantPartition(	×
233	known_args.partition_config_path)
234
235	beam_pipeline_options = pipeline_options.PipelineOptions(pipeline_args)	×
236	pipeline = beam.Pipeline(options=beam_pipeline_options)	×
237	variants = _read_variants(pipeline, known_args)	×
238	variants \|= 'FilterVariants' >> filter_variants.FilterVariants(	×
239	reference_names=known_args.reference_names)
240	if partitioner:	×
241	num_partitions = partitioner.get_num_partitions()	×
242	partitioned_variants = variants \| 'PartitionVariants' >> beam.Partition(	×
243	partition_variants.PartitionVariants(partitioner), num_partitions)
244	variants = []	×
245	for i in range(num_partitions):	×
246	if partitioner.should_keep_partition(i):	×
247	variants.append(partitioned_variants[i])	×
248	else:
249	num_partitions -= 1	×
250	else:
251	# By default we don't partition the data, so we have only 1 partition.
252	num_partitions = 1	×
253	variants = [variants]	×
254
255	for i in range(num_partitions):	×
256	if variant_merger:	×
257	variants[i] \|= ('MergeVariants' + str(i) >>	×
258	merge_variants.MergeVariants(variant_merger))
259	variants[i] \|= (	×
260	'ProcessVariants' + str(i) >>
261	beam.Map(processed_variant_factory.create_processed_variant).\
262	with_output_types(processed_variant.ProcessedVariant))
263	if partitioner and partitioner.should_flatten():	×
264	variants = [variants \| 'FlattenPartitions' >> beam.Flatten()]	×
265	num_partitions = 1	×
266
267	if known_args.output_table:	×
268	for i in range(num_partitions):	×
269	table_suffix = ''	×
270	if partitioner and partitioner.get_partition_name(i):	×
271	table_suffix = '_' + partitioner.get_partition_name(i)	×
272	table_name = known_args.output_table + table_suffix	×
273	_ = (variants[i] \| 'VariantToBigQuery' + table_suffix >>	×
274	variant_to_bigquery.VariantToBigQuery(
275	table_name,
276	header_fields,
277	variant_merger,
278	processed_variant_factory,
279	append=known_args.append,
280	update_schema_on_append=known_args.update_schema_on_append,
281	allow_incompatible_records=known_args.allow_incompatible_records,
282	omit_empty_sample_calls=known_args.omit_empty_sample_calls,
283	num_bigquery_write_shards=known_args.num_bigquery_write_shards,
284	null_numeric_value_replacement=(
285	known_args.null_numeric_value_replacement)))
286
287	if known_args.output_avro_path:	×
288	# TODO(bashir2): Add an integration test that outputs to Avro files and
289	# also imports to BigQuery. Then import those Avro outputs using the bq
290	# tool and verify that the two tables are identical.
291	_ = (	×
292	variants \| 'FlattenToOnePCollection' >> beam.Flatten()
293	\| 'VariantToAvro' >>
294	variant_to_avro.VariantToAvroFiles(
295	known_args.output_avro_path,
296	header_fields,
297	processed_variant_factory,
298	variant_merger=variant_merger,
299	allow_incompatible_records=known_args.allow_incompatible_records,
300	omit_empty_sample_calls=known_args.omit_empty_sample_calls,
301	null_numeric_value_replacement=(
302	known_args.null_numeric_value_replacement))
303	)
304
305	result = pipeline.run()	×
306	result.wait_until_finish()	×
307
308	metrics_util.log_all_counters(result)	×
309
310
311	if __name__ == '__main__':	1✔
312	logging.getLogger().setLevel(logging.INFO)	×
313	run()	×

googlegenomics / gcp-variant-transforms / 1528

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous