IBM Cloud Solution Architect (C1000-175) - Advanced Services & Integration¶

# watson_ai_integration.py
from ibm_watson import AssistantV2, DiscoveryV2, NaturalLanguageUnderstandingV1
from ibm_watson import SpeechToTextV1, TextToSpeechV1
from ibm_watson.natural_language_understanding_v1 import Features, EntitiesOptions, SentimentOptions
from ibm_cloud_sdk_core.authenticators import IAMAuthenticator
import json

class WatsonAIOrchestrator:
    """Orchestrate multiple Watson AI services"""

    def __init__(self, api_keys: dict, urls: dict):
        # Initialize Watson Assistant
        self.assistant = AssistantV2(
            version='2021-06-14',
            authenticator=IAMAuthenticator(api_keys['assistant'])
        )
        self.assistant.set_service_url(urls['assistant'])
        self.assistant_id = api_keys['assistant_id']

        # Initialize Watson Discovery
        self.discovery = DiscoveryV2(
            version='2020-08-30',
            authenticator=IAMAuthenticator(api_keys['discovery'])
        )
        self.discovery.set_service_url(urls['discovery'])
        self.project_id = api_keys['project_id']

        # Initialize NLU
        self.nlu = NaturalLanguageUnderstandingV1(
            version='2021-08-01',
            authenticator=IAMAuthenticator(api_keys['nlu'])
        )
        self.nlu.set_service_url(urls['nlu'])

    def intelligent_search(self, user_query: str):
        """Combine Assistant and Discovery for intelligent search"""
        # Step 1: Process query through Assistant
        session_response = self.assistant.create_session(
            assistant_id=self.assistant_id
        ).get_result()
        session_id = session_response['session_id']

        assistant_response = self.assistant.message(
            assistant_id=self.assistant_id,
            session_id=session_id,
            input={'message_type': 'text', 'text': user_query}
        ).get_result()

        # Step 2: Analyze sentiment and entities with NLU
        nlu_response = self.nlu.analyze(
            text=user_query,
            features=Features(
                entities=EntitiesOptions(sentiment=True, limit=10),
                sentiment=SentimentOptions()
            )
        ).get_result()

        # Step 3: Enhanced search with Discovery
        search_query = self._build_search_query(nlu_response)
        discovery_response = self.discovery.query(
            project_id=self.project_id,
            query=search_query
        ).get_result()

        # Step 4: Synthesize results
        return {
            'user_query': user_query,
            'intent': assistant_response.get('output', {}).get('intents', []),
            'sentiment': nlu_response['sentiment']['document'],
            'entities': nlu_response['entities'],
            'search_results': discovery_response['results'][:5],
            'answer': self._synthesize_answer(assistant_response, discovery_response)
        }

    def _build_search_query(self, nlu_response):
        """Build enhanced search query from NLU results"""
        entities = [e['text'] for e in nlu_response.get('entities', [])]
        return ' OR '.join(entities) if entities else ''

    def _synthesize_answer(self, assistant_resp, discovery_resp):
        """Synthesize answer from multiple sources"""
        assistant_text = assistant_resp.get('output', {}).get('generic', [])
        if assistant_text and assistant_text[0].get('text'):
            return assistant_text[0]['text']

        if discovery_resp.get('results'):
            return discovery_resp['results'][0].get('text', [''])[0][:200]

        return "I couldn't find a specific answer."

# Terraform for Watson services
watson_tf = '''
# watson-services.tf
resource "ibm_resource_instance" "watson_assistant" {
  name              = "production-assistant"
  service           = "conversation"
  plan              = "plus"
  location          = "us-south"
  resource_group_id = data.ibm_resource_group.default.id
}

resource "ibm_resource_instance" "watson_discovery" {
  name              = "production-discovery"
  service           = "discovery"
  plan              = "plus"
  location          = "us-south"
  resource_group_id = data.ibm_resource_group.default.id
}

resource "ibm_resource_instance" "watson_nlu" {
  name              = "production-nlu"
  service           = "natural-language-understanding"
  plan              = "standard"
  location          = "us-south"
  resource_group_id = data.ibm_resource_group.default.id
}

resource "ibm_resource_instance" "watson_speech_to_text" {
  name              = "production-stt"
  service           = "speech-to-text"
  plan              = "standard"
  location          = "us-south"
  resource_group_id = data.ibm_resource_group.default.id
}

# Service credentials
resource "ibm_resource_key" "watson_assistant_key" {
  name                 = "assistant-credentials"
  resource_instance_id = ibm_resource_instance.watson_assistant.id
  role                 = "Manager"
}

resource "kubernetes_secret" "watson_credentials" {
  metadata {
    name      = "watson-credentials"
    namespace = "production"
  }

  data = {
    assistant_apikey = ibm_resource_key.watson_assistant_key.credentials["apikey"]
    assistant_url    = ibm_resource_key.watson_assistant_key.credentials["url"]
    discovery_apikey = ibm_resource_key.watson_discovery_key.credentials["apikey"]
    discovery_url    = ibm_resource_key.watson_discovery_key.credentials["url"]
    nlu_apikey       = ibm_resource_key.watson_nlu_key.credentials["apikey"]
    nlu_url          = ibm_resource_key.watson_nlu_key.credentials["url"]
  }
}
'''

# analytics_pipeline.py
from ibm_watson import WatsonMachineLearningAPIClient
from ibm_botocore.client import Config
import ibm_boto3
import pandas as pd
from pyspark.sql import SparkSession
import json

class DataAnalyticsPipeline:
    """End-to-end data analytics pipeline on IBM Cloud"""

    def __init__(self, cos_config, wml_config):
        # Cloud Object Storage
        self.cos_client = ibm_boto3.client(
            's3',
            ibm_api_key_id=cos_config['api_key'],
            ibm_service_instance_id=cos_config['instance_id'],
            config=Config(signature_version='oauth'),
            endpoint_url=cos_config['endpoint']
        )

        # Watson Machine Learning
        self.wml_client = WatsonMachineLearningAPIClient(wml_config)

        # Spark session
        self.spark = SparkSession.builder \
            .appName("IBM Cloud Analytics") \
            .config("spark.hadoop.fs.s3a.endpoint", cos_config['endpoint']) \
            .config("spark.hadoop.fs.s3a.access.key", cos_config['access_key']) \
            .config("spark.hadoop.fs.s3a.secret.key", cos_config['secret_key']) \
            .getOrCreate()

    def ingest_data(self, source_bucket, source_key):
        """Ingest data from Cloud Object Storage"""
        # Read from COS
        obj = self.cos_client.get_object(
            Bucket=source_bucket,
            Key=source_key
        )

        # Load into Spark DataFrame
        df = self.spark.read.json(obj['Body'])
        return df

    def transform_data(self, df):
        """Transform data using Spark"""
        from pyspark.sql.functions import col, when, avg, count

        # Data quality checks
        df_clean = df.na.drop(how='any', subset=['customer_id', 'amount'])

        # Feature engineering
        df_features = df_clean.withColumn(
            'amount_category',
            when(col('amount') < 100, 'low')
            .when(col('amount') < 1000, 'medium')
            .otherwise('high')
        )

        # Aggregations
        df_agg = df_features.groupBy('customer_id').agg(
            count('*').alias('transaction_count'),
            avg('amount').alias('avg_amount')
        )

        return df_agg

    def store_curated_data(self, df, bucket, key):
        """Store curated data back to COS"""
        # Write as Parquet for efficiency
        df.write.mode('overwrite').parquet(
            f"s3a://{bucket}/{key}"
        )

    def train_model(self, training_data):
        """Train ML model with Watson Machine Learning"""
        # Define model metadata
        model_metadata = {
            "name": "Customer Segmentation Model",
            "type": "scikit-learn_1.0",
            "software_spec_uid": self.wml_client.software_specifications.get_uid_by_name("scikit-learn_1.0")
        }

        # Train model (example with scikit-learn)
        from sklearn.cluster import KMeans
        import pickle

        # Convert to pandas for sklearn
        pdf = training_data.toPandas()
        X = pdf[['transaction_count', 'avg_amount']]

        model = KMeans(n_clusters=3, random_state=42)
        model.fit(X)

        # Store model
        stored_model = self.wml_client.repository.store_model(
            model=model,
            meta_props=model_metadata,
            training_data=X
        )

        return stored_model.get('metadata', {}).get('id')

    def deploy_model(self, model_id):
        """Deploy model for scoring"""
        deployment_metadata = {
            "name": "Customer Segmentation Deployment",
            "online": {}
        }

        deployment = self.wml_client.deployments.create(
            model_id,
            meta_props=deployment_metadata
        )

        return deployment.get('metadata', {}).get('id')

    def run_pipeline(self, source_bucket, source_key, output_bucket, output_key):
        """Execute full pipeline"""
        print("1. Ingesting data...")
        df = self.ingest_data(source_bucket, source_key)

        print("2. Transforming data...")
        df_transformed = self.transform_data(df)

        print("3. Storing curated data...")
        self.store_curated_data(df_transformed, output_bucket, output_key)

        print("4. Training ML model...")
        model_id = self.train_model(df_transformed)

        print("5. Deploying model...")
        deployment_id = self.deploy_model(model_id)

        print(f"Pipeline complete! Deployment ID: {deployment_id}")
        return deployment_id

# Terraform for Analytics Infrastructure
analytics_tf = '''
# analytics-infrastructure.tf

# Analytics Engine (Spark cluster)
resource "ibm_analytics_engine" "spark_cluster" {
  name              = "analytics-spark-cluster"
  service_plan      = "standard"
  location          = "us-south"
  resource_group_id = data.ibm_resource_group.default.id
}

# Db2 Warehouse for analytics
resource "ibm_database" "db2_warehouse" {
  name              = "analytics-warehouse"
  plan              = "standard"
  location          = "us-south"
  service           = "dashdb-for-transactions"
  resource_group_id = data.ibm_resource_group.default.id

  group {
    group_id = "member"
    memory {
      allocation_mb = 16384
    }
    disk {
      allocation_mb = 102400
    }
    cpu {
      allocation_count = 6
    }
  }
}

# Watson Machine Learning
resource "ibm_resource_instance" "watson_ml" {
  name              = "production-watson-ml"
  service           = "pm-20"
  plan              = "v2-standard"
  location          = "us-south"
  resource_group_id = data.ibm_resource_group.default.id
}

# Cloud Object Storage for data lake
resource "ibm_resource_instance" "data_lake_cos" {
  name              = "data-lake-storage"
  service           = "cloud-object-storage"
  plan              = "standard"
  location          = "global"
  resource_group_id = data.ibm_resource_group.default.id
}

# COS buckets for different data layers
resource "ibm_cos_bucket" "raw_data" {
  bucket_name          = "raw-data-layer"
  resource_instance_id = ibm_resource_instance.data_lake_cos.id
  region_location      = "us-south"
  storage_class        = "standard"

  activity_tracking {
    read_data_events     = true
    write_data_events    = true
    activity_tracker_crn = ibm_resource_instance.activity_tracker.id
  }
}

resource "ibm_cos_bucket" "curated_data" {
  bucket_name          = "curated-data-layer"
  resource_instance_id = ibm_resource_instance.data_lake_cos.id
  region_location      = "us-south"
  storage_class        = "standard"
}

resource "ibm_cos_bucket" "analytics_data" {
  bucket_name          = "analytics-data-layer"
  resource_instance_id = ibm_resource_instance.data_lake_cos.id
  region_location      = "us-south"
  storage_class        = "smart"

  # Lifecycle policy for cost optimization
  archive_rule {
    rule_id = "archive-old-analytics"
    enable  = true
    days    = 90
    type    = "Glacier"
  }

  expire_rule {
    rule_id = "expire-old-data"
    enable  = true
    days    = 730  # 2 years
  }
}
'''

# cloud_functions_example.py
import json
import ibm_boto3
from ibm_botocore.client import Config
from ibm_watson import LanguageTranslatorV3
from ibm_cloud_sdk_core.authenticators import IAMAuthenticator

def main(args):
    """
    Cloud Function: Process uploaded document
    Triggered by Cloud Object Storage
    """
    # Get event details
    bucket = args.get('bucket')
    key = args.get('key')

    if not bucket or not key:
        return {
            'statusCode': 400,
            'body': json.dumps({'error': 'Missing bucket or key'})
        }

    try:
        # Initialize COS client
        cos = ibm_boto3.client(
            's3',
            ibm_api_key_id=args.get('cos_api_key'),
            ibm_service_instance_id=args.get('cos_instance_id'),
            config=Config(signature_version='oauth'),
            endpoint_url=args.get('cos_endpoint')
        )

        # Download file from COS
        obj = cos.get_object(Bucket=bucket, Key=key)
        content = obj['Body'].read().decode('utf-8')

        # Process with Watson
        translator = LanguageTranslatorV3(
            version='2018-05-01',
            authenticator=IAMAuthenticator(args.get('watson_api_key'))
        )
        translator.set_service_url(args.get('watson_url'))

        # Translate document
        translation = translator.translate(
            text=content,
            model_id='en-es'
        ).get_result()

        translated_text = translation['translations'][0]['translation']

        # Store result back to COS
        output_key = f"translated/{key}"
        cos.put_object(
            Bucket=bucket,
            Key=output_key,
            Body=translated_text.encode('utf-8')
        )

        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Document processed successfully',
                'output_location': f"{bucket}/{output_key}"
            })
        }

    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

# Deploy Cloud Function
deploy_script = '''
# deploy-function.sh

# Create action
ibmcloud fn action create process-document \
  --kind python:3.9 \
  cloud_functions_example.py \
  --memory 512 \
  --timeout 300000

# Bind parameters
ibmcloud fn action update process-document \
  --param cos_api_key "$COS_API_KEY" \
  --param cos_instance_id "$COS_INSTANCE_ID" \
  --param cos_endpoint "$COS_ENDPOINT" \
  --param watson_api_key "$WATSON_API_KEY" \
  --param watson_url "$WATSON_URL"

# Create COS trigger
ibmcloud fn trigger create cos-upload-trigger \
  --feed /whisk.system/cos/changes \
  --param bucket "documents" \
  --param event_types "write"

# Create rule to connect trigger and action
ibmcloud fn rule create process-on-upload \
  cos-upload-trigger \
  process-document
'''