class_docstr: Interact with Amazon SageMaker.
  
  Provide thick wrapper around
  :external+boto3:py:class:`boto3.client("sagemaker") <SageMaker.Client>`.
  
  Additional arguments (such as ``aws_conn_id``) may be specified and
  are passed down to the underlying AwsBaseHook.
  
  .. seealso::
      - :class:`airflow.providers.amazon.aws.hooks.base_aws.AwsBaseHook`
class_docstr_tok: ["Interact","with","Amazon","SageMaker",".","Provide","thick","wrapper","around",":","external+boto3",":","py",":","class",":","`","boto3.client","(","``","sagemaker","''",")","<","SageMaker.Client",">","`",".","Additional","arguments","(","such","as","``","aws_conn_id","``",")","may","be","specified","and","are","passed","down","to","the","underlying","AwsBaseHook",".","..","seealso",":",":","-",":","class",":","`","airflow.providers.amazon.aws.hooks.base_aws.AwsBaseHook","`"]
docstr: Get the associated training job info and print CloudWatch logs.
code: def describe_training_job_with_log(
      self,
      job_name: str,
      positions,
      stream_names: list,
      instance_count: int,
      state: int,
      last_description: dict,
      last_describe_job_call: float,
  ):
      """Get the associated training job info and print CloudWatch logs."""
      log_group = "/aws/sagemaker/TrainingJobs"
  
      if len(stream_names) < instance_count:
          # Log streams are created whenever a container starts writing to stdout/err, so this list
          # may be dynamic until we have a stream for every instance.
          logs_conn = self.logs_hook.get_conn()
          try:
              streams = logs_conn.describe_log_streams(
                  logGroupName=log_group,
                  logStreamNamePrefix=job_name + "/",
                  orderBy="LogStreamName",
                  limit=instance_count,
              )
              stream_names = [
                  s["logStreamName"] for s in streams["logStreams"]
              ]
              positions.update(
                  [
                      (s, Position(timestamp=0, skip=0))
                      for s in stream_names
                      if s not in positions
                  ]
              )
          except logs_conn.exceptions.ResourceNotFoundException:
              # On the very first training job run on an account, there's no log group until
              # the container starts logging, so ignore any errors thrown about that
              pass
  
      if stream_names:
          for idx, event in self.multi_stream_iter(
              log_group, stream_names, positions
          ):
              self.log.info(event["message"])
              ts, count = positions[stream_names[idx]]
              if event["timestamp"] == ts:
                  positions[stream_names[idx]] = Position(
                      timestamp=ts, skip=count + 1
                  )
              else:
                  positions[stream_names[idx]] = Position(
                      timestamp=event["timestamp"], skip=1
                  )
  
      if state == LogState.COMPLETE:
          return state, last_description, last_describe_job_call
  
      if state == LogState.JOB_COMPLETE:
          state = LogState.COMPLETE
      elif time.monotonic() - last_describe_job_call >= 30:
          description = self.describe_training_job(job_name)
          last_describe_job_call = time.monotonic()
  
          if secondary_training_status_changed(
              description, last_description
          ):
              self.log.info(
                  secondary_training_status_message(
                      description, last_description
                  )
              )
              last_description = description
  
          status = description["TrainingJobStatus"]
  
          if status not in self.non_terminal_states:
              state = LogState.JOB_COMPLETE
      return state, last_description, last_describe_job_call
  
imports: from __future__ import annotations
  import os
  import re
  import tarfile
  import tempfile
  import time
  import warnings
  from collections import Counter, namedtuple
  from datetime import datetime
  from functools import partial
  from typing import Any, Callable, Generator, cast
  from botocore.exceptions import ClientError
  from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
  from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook
  from airflow.providers.amazon.aws.hooks.logs import AwsLogsHook
  from airflow.providers.amazon.aws.hooks.s3 import S3Hook
  from airflow.providers.amazon.aws.utils.tags import format_tags
  from airflow.utils import timezone
label_desc: Use image node_id 17 for calling the SageMakerHook obj's underlying member method code with example usage: obj.describe_training_job_with_log(job_name, positions, stream_names, instance_count, state, last_description, last_describe_job_call) and returns: state,  last_description,  last_describe_job_call, state,  last_description,  last_describe_job_call