debug: timeout fix

* Use sample dataset from regional bucket

* retrigger checks

.cloud-build/execute_changed_notebooks_helper.py

@@ -245,7 +245,7 @@ def process_and_execute_notebook(
         result.logs_bucket = operation_metadata.build.logs_bucket
 
         # Block and wait for the result
-        operation_result = operation.result()
+        operation_result = operation.result(timeout=86400)
 
         result.duration = datetime.datetime.now() - time_start
         result.is_pass = True

community-content/pytorch_efficient_training/README.md

@@ -15,15 +15,19 @@ pip install -r requirements.txt
 * resnet_dp.py - Train ResNet-50 on single node multiple GPUs with `DataParallel` strategy.
 * resnet_ddp.py - Train ResNet-50 on single node multiple GPUs with `DistributedDataParallel` strategy.
 * resnet_ddp_wds.py - Train ResNet-50 on single node multiple GPUs with `DistributedDataParallel` strategy and `Webdataset`.
+* resnet_fsdp.py - Train ResNet-50 on single node multiple GPUs with `FullyShardedDataParallel` strategy.
+* resnet_fsdp_wds.py - Train ResNet-50 on single node multiple GPUs with `FullyShardedDataParallel` strategy and `Webdataset`.
 * shard_imagenet.py - Shard ImagNet individual files into `tar` files.
 
 ## Benchmark
 
 When run the benchmark on Nvidia T4 GPUs using ImageNet validation dataset, you can get the result like:
-Strategy              | Seconds/Epoch - Local Data | Seconds/Epoch - Cloud Data
---------------------- | -------------------------- | --------------------------
-On 1 GPU              | 489                        | 804 (2x slower)
-On 4 GPUs (DP)        | 157                        | 738 (5x slower)
-On 4 GPUs (DDP)       | 134                        | 432 (3x slower)
-On 4 GPUs (DDP + WDS) | 131                        | 133 (same performance)
+Strategy               | Seconds/Epoch - Local Data | Seconds/Epoch - Cloud Data
+---------------------- | -------------------------- | --------------------------
+On 1 GPU               | 489                        | 804 (2x slower)
+On 4 GPUs (DP)         | 157                        | 738 (5x slower)
+On 4 GPUs (DDP)        | 134                        | 432 (3x slower)
+On 4 GPUs (DDP + WDS)  | 131                        | 133 (same performance)
+On 4 GPUs (FSDP)       | 139                        | 353 (3x slower)
+On 4 GPUs (FSDP + WDS) | 138                        | 135 (same performance)
 

community-content/pytorch_efficient_training/resnet_fsdp.py

@@ -0,0 +1,242 @@
+# Copyright 2022 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the \"License\");
+# you may not use this file except in compliance with the License.\n",
+# 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.
+
+"""Train resnet on multiple GPUs with FSDP."""
+
+import argparse
+import functools
+import os
+import time
+
+from PIL import Image
+import torch
+from torch import nn
+import torch.distributed as dist
+from torch.distributed.fsdp import FullyShardedDataParallel as FSDP
+from torch.distributed.fsdp.wrap import size_based_auto_wrap_policy
+import torch.multiprocessing as mp
+import torchmetrics
+import torchvision
+from torchvision.models import resnet50
+
+
+class ImageFolder(torchvision.datasets.ImageFolder):
+  """Class for loading imagenet."""
+
+  def __init__(self, image_list_file, transform=None, target_transform=None):
+    self.samples = self._make_dataset(image_list_file)
+    self.loader = self._loader
+
+    self.imgs = self.samples
+    self.targets = [s[1] for s in self.samples]
+
+    self.transform = transform
+    self.target_transform = target_transform
+
+  def _make_dataset(self, image_list_file):
+    items = []
+    with open(image_list_file, 'r') as f:
+      for line in f:
+        item = line.strip().split(' ')
+        items.append((item[0], int(item[1])))
+    return items
+
+  def _loader(self, image_path):
+    with open(image_path, 'rb') as f:
+      img = Image.open(f)
+      img = img.convert('RGB')
+      return img
+
+
+def train(model, device, dataloader, optimizer):
+  model.train()
+  for image, target in dataloader:
+    image = image.to(device, non_blocking=True)
+    target = target.to(device, non_blocking=True)
+    pred = model(image)
+    # pred.shape (N, C), target.shape (N)
+    loss = nn.functional.cross_entropy(pred, target)
+    optimizer.zero_grad()
+    loss.backward()
+    optimizer.step()
+  return loss
+
+
+def evaluate(model, device, dataloader, metric):
+  model.eval()
+  with torch.no_grad():
+    for image, target in dataloader:
+      image = image.to(device, non_blocking=True)
+      target = target.to(device, non_blocking=True)
+      pred = model(image)
+      metric.update(pred, target)
+  accuracy = metric.compute()
+  metric.reset()
+  return accuracy
+
+
+def worker(gpu, args):
+  """Run training and evaluation."""
+  # Init process group.
+  print(f'Initiating process {gpu}')
+  dist.init_process_group(
+      backend='nccl',
+      init_method='env://',
+      world_size=args.gpus,
+      rank=gpu)
+
+  # Create train dataloader.
+  train_dataset = ImageFolder(
+      image_list_file=args.train_data_path,
+      transform=torchvision.transforms.Compose([
+          torchvision.transforms.RandomResizedCrop(224),
+          torchvision.transforms.RandomHorizontalFlip(),
+          torchvision.transforms.ToTensor(),
+          torchvision.transforms.Normalize(
+              mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
+      ]))
+  train_sampler = torch.utils.data.distributed.DistributedSampler(
+      train_dataset, num_replicas=args.gpus, rank=gpu)
+  train_dataloader = torch.utils.data.DataLoader(
+      dataset=train_dataset,
+      batch_size=args.train_batch_size,
+      shuffle=False,
+      num_workers=args.dataloader_num_workers,
+      pin_memory=True,
+      sampler=train_sampler)
+  if gpu == 0:
+    print(f'Train dataloader | samples: {len(train_dataloader.dataset)}, '
+          f'num workers: {train_dataloader.num_workers}, '
+          f'global batch size: {args.train_batch_size * args.gpus}, '
+          f'batches/epoch: {len(train_dataloader)}')
+
+  # Create eval dataloader.
+  eval_dataset = ImageFolder(
+      image_list_file=args.eval_data_path,
+      transform=torchvision.transforms.Compose([
+          torchvision.transforms.Resize(256),
+          torchvision.transforms.CenterCrop(224),
+          torchvision.transforms.ToTensor(),
+          torchvision.transforms.Normalize(
+              mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
+      ]))
+  eval_sampler = torch.utils.data.distributed.DistributedSampler(
+      eval_dataset, num_replicas=args.gpus, rank=gpu)
+  eval_dataloader = torch.utils.data.DataLoader(
+      dataset=eval_dataset,
+      batch_size=args.eval_batch_size,
+      shuffle=False,
+      num_workers=args.dataloader_num_workers,
+      pin_memory=True,
+      drop_last=True,
+      sampler=eval_sampler)
+  if gpu == 0:
+    print(f'Eval dataloader | samples: {len(eval_dataloader.dataset)}, '
+          f'num workers: {eval_dataloader.num_workers}, '
+          f'batch size: {args.eval_batch_size}, '
+          f'batches/epoch: {len(eval_dataloader)}')
+
+  # Wrap policy.
+  my_auto_wrap_policy = functools.partial(
+      size_based_auto_wrap_policy, min_num_params=100)
+  torch.cuda.set_device(gpu)
+
+  # Create model.
+  model = resnet50(weights=None)
+  model.to(args.device)
+  model = FSDP(model, auto_wrap_policy=my_auto_wrap_policy)
+
+  # Optimizer.
+  optimizer = torch.optim.SGD(model.parameters(), 0.1)
+
+  # Main loop.
+  metric = torchmetrics.classification.Accuracy(top_k=1).to(args.device)
+  for epoch in range(1, args.epochs + 1):
+    if gpu == 0:
+      print(f'Running epoch {epoch}')
+    train_sampler.set_epoch(epoch)
+
+    start = time.time()
+    train(model, args.device, train_dataloader, optimizer)
+    end = time.time()
+    if gpu == 0:
+      print(f'Training finished in {(end - start):>0.3f} seconds')
+
+    start = time.time()
+    evaluate(model, args.device, eval_dataloader, metric)
+    end = time.time()
+    if gpu == 0:
+      print(f'Evaluation finished in {(end - start):>0.3f} seconds')
+
+  if gpu == 0:
+    print('Done')
+  dist.destroy_process_group()
+
+
+def create_args():
+  """Create main args."""
+  parser = argparse.ArgumentParser(
+      formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+  parser.add_argument(
+      '--gpus',
+      default=4,
+      type=int,
+      help='number of gpus to use')
+  parser.add_argument(
+      '--epochs',
+      default=2,
+      type=int,
+      help='number of total epochs to run')
+  parser.add_argument(
+      '--dataloader_num_workers',
+      default=2,
+      type=int,
+      help='number of workders for dataloader')
+  parser.add_argument(
+      '--train_data_path',
+      default='',
+      type=str,
+      help='path to training data')
+  parser.add_argument(
+      '--train_batch_size',
+      default=32,
+      type=int,
+      help='batch size for training per gpu')
+  parser.add_argument(
+      '--eval_data_path',
+      default='',
+      type=str,
+      help='path to evaluation data')
+  parser.add_argument(
+      '--eval_batch_size',
+      default=32,
+      type=int,
+      help='batch size for evaluation per gpu')
+  args = parser.parse_args()
+  return args
+
+
+def main():
+  args = create_args()
+
+  os.environ['MASTER_ADDR'] = 'localhost'
+  os.environ['MASTER_PORT'] = '8888'
+
+  args.device = 'cuda' if torch.cuda.is_available() else 'cpu'
+  print(f'Launch job on {args.gpus} GPUs with FSDP')
+  mp.spawn(worker, nprocs=args.gpus, args=(args,))
+
+
+if __name__ == '__main__':
+  main()

community-content/pytorch_efficient_training/resnet_fsdp_wds.py

@@ -0,0 +1,240 @@
+"""Train resnet on multiple GPUs with DDP."""
+
+import argparse
+import functools
+import itertools
+import math
+import os
+import time
+
+import torch
+from torch import nn
+import torch.distributed as dist
+from torch.distributed.fsdp import FullyShardedDataParallel as FSDP
+from torch.distributed.fsdp.wrap import size_based_auto_wrap_policy
+import torch.multiprocessing as mp
+import torchmetrics
+from torchvision.models import resnet50
+from torchvision.transforms import transforms
+import webdataset as wds
+
+
+def wds_split(src, rank, world_size):
+  """Shards split function for webdataset."""
+  # The context of caller of this function is within multiple processes
+  # (by DDP world_size) and multiple workers (by dataloader_num_workers).
+  # So we totally have (world_size * num_workers) workers for processing data.
+  # NOTE: Raw data should be sharded to enough shards to make sure one process
+  # can handle at least one shard, otherwise the process may hang.
+  worker_id = 0
+  num_workers = 1
+  worker_info = torch.utils.data.get_worker_info()
+  if worker_info:
+    worker_id = worker_info.id
+    num_workers = worker_info.num_workers
+  for s in itertools.islice(src, rank * num_workers + worker_id, None,
+                            world_size * num_workers):
+    yield s
+
+
+def identity(x):
+  return x
+
+
+def create_wds_dataloader(rank, args, mode):
+  """Create webdataset dataset and dataloader."""
+  if mode == 'train':
+    transform = transforms.Compose([
+        transforms.RandomResizedCrop(224),
+        transforms.RandomHorizontalFlip(),
+        transforms.ToTensor(),
+        transforms.Normalize(
+            mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
+    ])
+    data_path = args.train_data_path
+    data_size = args.train_data_size
+    batch_size_local = args.train_batch_size
+    batch_size_global = args.train_batch_size * args.gpus
+    # Since webdataset disallows partial batch, we pad the last batch for train.
+    batches = int(math.ceil(data_size / batch_size_global))
+  else:
+    transform = transforms.Compose([
+        transforms.Resize(256),
+        transforms.CenterCrop(224),
+        transforms.ToTensor(),
+        transforms.Normalize(
+            mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
+    ])
+    data_path = args.eval_data_path
+    data_size = args.eval_data_size
+    batch_size_local = args.eval_batch_size
+    batch_size_global = args.eval_batch_size * args.gpus
+    # Since webdataset disallows partial batch, we drop the last batch for eval.
+    batches = int(data_size / batch_size_global)
+
+  dataset = wds.DataPipeline(
+      wds.SimpleShardList(data_path),
+      functools.partial(wds_split, rank=rank, world_size=args.gpus),
+      wds.tarfile_to_samples(),
+      wds.decode('pil'),
+      wds.to_tuple('jpg;png;jpeg cls'),
+      wds.map_tuple(transform, identity),
+      wds.batched(batch_size_local, partial=False),
+  )
+  num_workers = args.dataloader_num_workers
+  dataloader = wds.WebLoader(
+      dataset=dataset,
+      batch_size=None,
+      shuffle=False,
+      num_workers=num_workers,
+      persistent_workers=True if num_workers > 0 else False,
+      pin_memory=True).repeat(nbatches=batches)
+  print(f'{mode} dataloader | samples: {data_size}, '
+        f'num_workers: {num_workers}, '
+        f'local batch size: {batch_size_local}, '
+        f'global batch size: {batch_size_global}, '
+        f'batches: {batches}')
+  return dataloader
+
+
+def train(model, device, dataloader, optimizer):
+  model.train()
+  for image, target in dataloader:
+    image = image.to(device, non_blocking=True)
+    target = target.to(device, non_blocking=True)
+    pred = model(image)
+    # pred.shape (N, C), target.shape (N)
+    loss = nn.functional.cross_entropy(pred, target)
+    optimizer.zero_grad()
+    loss.backward()
+    optimizer.step()
+  return loss
+
+
+def evaluate(model, device, dataloader, metric):
+  model.eval()
+  with torch.no_grad():
+    for image, target in dataloader:
+      image = image.to(device, non_blocking=True)
+      target = target.to(device, non_blocking=True)
+      pred = model(image)
+      metric.update(pred, target)
+  accuracy = metric.compute()
+  metric.reset()
+  return accuracy
+
+
+def worker(gpu, args):
+  """Run training and evaluation."""
+  # Init process group.
+  print(f'Initiating process {gpu}')
+  dist.init_process_group(
+      backend='nccl',
+      init_method='env://',
+      world_size=args.gpus,
+      rank=gpu)
+
+  # Create dataloader.
+  train_dataloader = create_wds_dataloader(gpu, args, 'train')
+  eval_dataloader = create_wds_dataloader(gpu, args, 'eval')
+
+  # Wrap policy.
+  my_auto_wrap_policy = functools.partial(
+      size_based_auto_wrap_policy, min_num_params=100)
+  torch.cuda.set_device(gpu)
+
+  # Create model.
+  model = resnet50(weights=None)
+  model.to(args.device)
+  model = FSDP(model, auto_wrap_policy=my_auto_wrap_policy)
+
+  # Optimizer.
+  optimizer = torch.optim.SGD(model.parameters(), 0.1)
+
+  # Main loop.
+  metric = torchmetrics.classification.Accuracy(top_k=1).to(args.device)
+  for epoch in range(1, args.epochs + 1):
+    if gpu == 0:
+      print(f'Running epoch {epoch}')
+
+    start = time.time()
+    train(model, args.device, train_dataloader, optimizer)
+    end = time.time()
+    if gpu == 0:
+      print(f'Training finished in {(end - start):>0.3f} seconds')
+
+    start = time.time()
+    evaluate(model, args.device, eval_dataloader, metric)
+    end = time.time()
+    if gpu == 0:
+      print(f'Evaluation finished in {(end - start):>0.3f} seconds')
+
+  if gpu == 0:
+    print('Done')
+
+
+def create_args():
+  """Create main args."""
+  parser = argparse.ArgumentParser(
+      formatter_class=argparse.ArgumentDefaultsHelpFormatter)
+  parser.add_argument(
+      '--gpus',
+      default=4,
+      type=int,
+      help='number of gpus to use')
+  parser.add_argument(
+      '--epochs',
+      default=2,
+      type=int,
+      help='number of total epochs to run')
+  parser.add_argument(
+      '--dataloader_num_workers',
+      default=2,
+      type=int,
+      help='number of workders for dataloader')
+  parser.add_argument(
+      '--train_data_path',
+      default='',
+      type=str,
+      help='path to training data')
+  parser.add_argument(
+      '--train_batch_size',
+      default=32,
+      type=int,
+      help='batch size for training per gpu')
+  parser.add_argument(
+      '--train_data_size',
+      default=50000,
+      type=int,
+      help='data size for training')
+  parser.add_argument(
+      '--eval_data_path',
+      default='',
+      type=str,
+      help='path to evaluation data')
+  parser.add_argument(
+      '--eval_batch_size',
+      default=32,
+      type=int,
+      help='batch size for evaluation per gpu')
+  parser.add_argument(
+      '--eval_data_size',
+      default=50000,
+      type=int,
+      help='data size for evaluation')
+  args = parser.parse_args()
+  return args
+
+
+def main():
+  args = create_args()
+  os.environ['MASTER_ADDR'] = 'localhost'
+  os.environ['MASTER_PORT'] = '8888'
+
+  args.device = 'cuda' if torch.cuda.is_available() else 'cpu'
+  print(f'Launch job on {args.gpus} GPUs with FSDP')
+  mp.spawn(worker, nprocs=args.gpus, args=(args,))
+
+
+if __name__ == '__main__':
+  main()

notebooks/community/ml_ops/stage5/get_started_with_vertex_endpoint_and_shared_vm.ipynb

@@ -159,9 +159,9 @@
         "\n",
         "# Install the packages\n",
         "\n",
-        "! pip3 install --upgrade google-cloud-aiplatform $USER_FLAG -q\n",
-        "! pip3 install --upgrade tensorflow $USER_FLAG -q\n",
-        "! pip3 install --upgrade tensorflow-hub $USER_FLAG -q"
+        "! pip3 install --upgrade google-cloud-aiplatform \\\n",
+        "                         tensorflow \\\n",
+        "                         tensorflow-hub $USER_FLAG -q"
       ]
     },
     {
@@ -307,22 +307,29 @@
         "id": "timestamp"
       },
       "source": [
-        "#### Timestamp\n",
+        "#### UUID\n",
         "\n",
-        "If you are in a live tutorial session, you might be using a shared test account or project. To avoid name collisions between users on resources created, you create a timestamp for each instance session, and append the timestamp onto the name of resources you create in this tutorial."
+        "If you are in a live tutorial session, you might be using a shared test account or project. To avoid name collisions between users on resources created, you create a uuid for each instance session, and append it onto the name of resources you create in this tutorial."
       ]
     },
     {
       "cell_type": "code",
       "execution_count": null,
       "metadata": {
-        "id": "timestamp"
+        "id": "84Vdv7R-QEH6"
       },
       "outputs": [],
       "source": [
-        "from datetime import datetime\n",
+        "import random\n",
+        "import string\n",
         "\n",
-        "TIMESTAMP = datetime.now().strftime(\"%Y%m%d%H%M%S\")"
+        "\n",
+        "# Generate a uuid of a specifed length(default=8)\n",
+        "def generate_uuid(length: int = 8) -> str:\n",
+        "    return \"\".join(random.choices(string.ascii_lowercase + string.digits, k=length))\n",
+        "\n",
+        "\n",
+        "UUID = generate_uuid()"
       ]
     },
     {
@@ -421,7 +428,7 @@
       "outputs": [],
       "source": [
         "if BUCKET_URI == \"\" or BUCKET_URI is None or BUCKET_URI == \"gs://[your-bucket-name]\":\n",
-        "    BUCKET_NAME = PROJECT_ID + \"aip-\" + TIMESTAMP\n",
+        "    BUCKET_NAME = PROJECT_ID + \"aip-\" + UUID\n",
         "    BUCKET_URI = \"gs://\" + BUCKET_NAME"
       ]
     },
@@ -523,7 +530,7 @@
         "\n",
         "Setup up the following constants for Vertex AI:\n",
         "\n",
-        "- `API_ENDPOINT`: The Vertex AI API service endpoint for `Endpoint` services."
+        "- `API_ENDPOINT`: The Vertex AI API service endpoint."
       ]
     },
     {
@@ -538,46 +545,10 @@
         "API_ENDPOINT = \"{}-aiplatform.googleapis.com\".format(REGION)\n",
         "\n",
         "# Vertex location root path for your dataset, model and endpoint resources\n",
-        "PARENT = \"projects/\" + PROJECT_ID + \"/locations/\" + REGION"
-      ]
-    },
-    {
-      "cell_type": "markdown",
-      "metadata": {
-        "id": "clients:metadata"
-      },
-      "source": [
-        "## Set up clients\n",
+        "PARENT = \"projects/\" + PROJECT_ID + \"/locations/\" + REGION\n",
         "\n",
-        "The Vertex  works as a client/server model. On your side (the Python script) you will create a client that sends requests and receives responses from the Vertex AI server.\n",
-        "\n",
-        "You will use different clients in this tutorial for different steps in the workflow. So set them all up upfront.\n",
-        "\n",
-        "- Endpoint Service for creating endpoints, and deploying models to endpoints."
-      ]
-    },
-    {
-      "cell_type": "code",
-      "execution_count": null,
-      "metadata": {
-        "id": "clients:metadata"
-      },
-      "outputs": [],
-      "source": [
         "# client options same for all services\n",
-        "client_options = {\"api_endpoint\": API_ENDPOINT}\n",
-        "\n",
-        "\n",
-        "def create_endpoint_client():\n",
-        "    client = aip_beta.EndpointServiceClient(client_options=client_options)\n",
-        "    return client\n",
-        "\n",
-        "\n",
-        "clients = {}\n",
-        "clients[\"endpoint\"] = create_endpoint_client()\n",
-        "\n",
-        "for client in clients.items():\n",
-        "    print(client)"
+        "client_options = {\"api_endpoint\": API_ENDPOINT}"
       ]
     },
     {
@@ -592,7 +563,7 @@
         "\n",
         "Set the variables `DEPLOY_GPU/DEPLOY_NGPU` to use a container image supporting a GPU and the number of GPUs allocated to the virtual machine (VM) instance. For example, to use a GPU container image with 4 Nvidia Telsa K80 GPUs allocated to each VM, you would specify:\n",
         "\n",
-        "    (aip.AcceleratorType.NVIDIA_TESLA_K80, 4)\n",
+        "    (aip.gapic.AcceleratorType.NVIDIA_TESLA_K80, 4)\n",
         "\n",
         "\n",
         "Otherwise specify `(None, None)` to use a container image to run on a CPU.\n",
@@ -902,7 +873,7 @@
       "outputs": [],
       "source": [
         "model_icn = aiplatform.Model.upload(\n",
-        "    display_name=\"icn_\" + TIMESTAMP,\n",
+        "    display_name=\"icn_\" + UUID,\n",
         "    artifact_uri=MODEL_ICN_DIR,\n",
         "    serving_container_image_uri=DEPLOY_IMAGE,\n",
         ")\n",
@@ -1013,7 +984,7 @@
       "outputs": [],
       "source": [
         "model_use = aiplatform.Model.upload(\n",
-        "    display_name=\"icn_\" + TIMESTAMP,\n",
+        "    display_name=\"icn_\" + UUID,\n",
         "    artifact_uri=MODEL_USE_DIR,\n",
         "    serving_container_image_uri=DEPLOY_IMAGE,\n",
         ")\n",
@@ -1029,64 +1000,52 @@
       "source": [
         "## Creating a deployment resource pool\n",
         "\n",
-        "Currently, creating deploynent resource pools is only supported via the REST-based API (e.g., CURL).\n",
+        "Currently, creating deploynent resource pools is only supported via the REST-based API (e.g., CURL) and GAPIC APIs (Python).\n",
         "\n",
-        "Use `CreateDeploymentResourcePool` API to create a resource pool, with the following configuration:\n",
+        "Use `create_deployment_resource_pool` API to create a resource pool, with the following configuration:\n",
         "\n",
         "- `dedicated_resources`: Compute (HW) resources to allocate for the shared vm.\n",
         "- `min_replica_count`: Auto-scaling, the minimum number of compute nodes.\n",
         "- `max_replica_count`: Auto-scaling, the maximum number of compute nodes.\n",
         "\n",
-        "Learn more about [Deployment Resource Pools]()."
+        "Learn more about [Deployment Resource Pools](https://googleapis.dev/python/aiplatform/latest/aiplatform_v1beta1/deployment_resource_pool_service.html)."
       ]
     },
     {
       "cell_type": "code",
       "execution_count": null,
       "metadata": {
-        "id": "YiBmoiWYcMQt"
+        "id": "90c51b6cf34a"
       },
       "outputs": [],
       "source": [
-        "DEPLOYMENT_RESOURCE_POOL_ID = \"shared-vm\"  # @param {type: \"string\"}"
-      ]
-    },
-    {
-      "cell_type": "code",
-      "execution_count": null,
-      "metadata": {
-        "id": "0CHPJ4h-Slgs"
-      },
-      "outputs": [],
-      "source": [
-        "import json\n",
-        "import pprint\n",
-        "pp = pprint.PrettyPrinter(indent=4)\n",
-        "\n",
+        "DEPLOYMENT_RESOURCE_POOL_ID = f\"shared-vm-{UUID}\"  # @param {type: \"string\"}\n",
         "MIN_NODES = 1\n",
         "MAX_NODES = 2\n",
         "\n",
-        "CREATE_RP_PAYLOAD = {\n",
-        "  \"deployment_resource_pool\":{\n",
-        "    \"dedicated_resources\":{\n",
-        "      \"machine_spec\":{\n",
-        "        \"machine_type\": DEPLOY_COMPUTE\n",
-        "      },\n",
-        "      \"min_replica_count\": MIN_NODES, \n",
-        "      \"max_replica_count\": MAX_NODES\n",
-        "    }\n",
-        "  },\n",
-        "  \"deployment_resource_pool_id\":DEPLOYMENT_RESOURCE_POOL_ID\n",
-        "}\n",
-        "CREATE_RP_REQUEST=json.dumps(CREATE_RP_PAYLOAD)\n",
-        "pp.pprint(\"CREATE_RP_REQUEST: \" + CREATE_RP_REQUEST)\n",
+        "# Initialize request argument(s)\n",
+        "deployment_resource_pool = aip_beta.DeploymentResourcePool()\n",
+        "deployment_resource_pool.dedicated_resources.min_replica_count = MIN_NODES\n",
+        "deployment_resource_pool.dedicated_resources.max_replica_count = MAX_NODES\n",
+        "deployment_resource_pool.dedicated_resources.machine_spec.machine_type = DEPLOY_COMPUTE\n",
         "\n",
-        "! curl \\\n",
-        "-X POST \\\n",
-        "-H \"Authorization: Bearer $(gcloud auth print-access-token)\" \\\n",
-        "-H \"Content-Type: application/json\" \\\n",
-        "https://{REGION}-aiplatform.googleapis.com/v1beta1/projects/{PROJECT_ID}/locations/{REGION}/deploymentResourcePools \\\n",
-        "-d '{CREATE_RP_REQUEST}'"
+        "request = aip_beta.CreateDeploymentResourcePoolRequest(\n",
+        "    parent=f\"projects/{PROJECT_ID}/locations/{REGION}\",\n",
+        "    deployment_resource_pool=deployment_resource_pool,\n",
+        "    deployment_resource_pool_id=DEPLOYMENT_RESOURCE_POOL_ID,\n",
+        ")\n",
+        "\n",
+        "pool_client = aip_beta.services.deployment_resource_pool_service.DeploymentResourcePoolServiceClient(\n",
+        "    client_options=client_options\n",
+        ")\n",
+        "\n",
+        "op = pool_client.create_deployment_resource_pool(request=request)\n",
+        "print(op)\n",
+        "\n",
+        "result = op.result()\n",
+        "print(result)\n",
+        "\n",
+        "deployment_pool_id = result.name"
       ]
     },
     {
@@ -1099,21 +1058,19 @@
         "\n",
         "Use `GetDeploymentResourcePool` API to check out the deploynent resource pool that you created. \n",
         "\n",
-        "Learn more about [Get Deployment Resource Pool](https://source.corp.google.com/piper///depot/google3/google/cloud/aiplatform/master/deployment_resource_pool_service.proto;l=75?q=deployment_resource_pool&sq=package:piper%20file:%2F%2Fdepot%2Fgoogle3%20-file:google3%2Fexperimental)."
+        "Learn more about [Get Deployment Resource Pool](https://googleapis.dev/python/aiplatform/latest/aiplatform_v1beta1/deployment_resource_pool_service.html)."
       ]
     },
     {
       "cell_type": "code",
       "execution_count": null,
       "metadata": {
-        "id": "6wTLyhPraFah"
+        "id": "b740253903c0"
       },
       "outputs": [],
       "source": [
-        "! curl -X GET \\\n",
-        "-H \"Authorization: Bearer $(gcloud auth print-access-token)\" \\\n",
-        "-H \"Content-Type: application/json\" \\\n",
-        "https://{REGION}-aiplatform.googleapis.com/v1beta1/projects/{PROJECT_ID}/locations/{REGION}/deploymentResourcePools/{DEPLOYMENT_RESOURCE_POOL_ID}"
+        "response = pool_client.get_deployment_resource_pool(name=deployment_pool_id)\n",
+        "print(response)"
       ]
     },
     {
@@ -1126,21 +1083,22 @@
         "\n",
         "Use `ListDeploymentResourcePools` API to list all the deployment resource pools. \n",
         "\n",
-        "Learn more about [Listing Deployment Resource Pools](https://source.corp.google.com/piper///depot/google3/google/cloud/aiplatform/master/deployment_resource_pool_service.proto;l=101?q=deployment_resource_pool&sq=package:piper%20file:%2F%2Fdepot%2Fgoogle3%20-file:google3%2Fexperimental)."
+        "Learn more about [Listing Deployment Resource Pools](https://googleapis.dev/python/aiplatform/latest/aiplatform_v1beta1/deployment_resource_pool_service.html)."
       ]
     },
     {
       "cell_type": "code",
       "execution_count": null,
       "metadata": {
-        "id": "Pxls4sNnaltU"
+        "id": "3ebfd007bff2"
       },
       "outputs": [],
       "source": [
-        "! curl -X GET \\\n",
-        "-H \"Authorization: Bearer $(gcloud auth print-access-token)\" \\\n",
-        "-H \"Content-Type: application/json\" \\\n",
-        "https://{REGION}-aiplatform.googleapis.com/v1beta1/projects/{PROJECT_ID}/locations/{REGION}/deploymentResourcePools"
+        "pools = pool_client.list_deployment_resource_pools(\n",
+        "    parent=f\"projects/{PROJECT_ID}/locations/{REGION}\"\n",
+        ")\n",
+        "for pool in pools:\n",
+        "    print(pool)"
       ]
     },
     {
@@ -1170,11 +1128,11 @@
       },
       "outputs": [],
       "source": [
-        "endpoint_icn = aiplatform.Endpoint.create(display_name=\"icn_\" + TIMESTAMP)\n",
+        "endpoint_icn = aiplatform.Endpoint.create(display_name=\"icn_\" + UUID)\n",
         "\n",
         "print(endpoint_icn)\n",
         "\n",
-        "endpoint_use = aiplatform.Endpoint.create(display_name=\"use_\" + TIMESTAMP)\n",
+        "endpoint_use = aiplatform.Endpoint.create(display_name=\"use_\" + UUID)\n",
         "\n",
         "print(endpoint_use)"
       ]
@@ -1204,6 +1162,12 @@
       },
       "outputs": [],
       "source": [
+        "import json\n",
+        "import pprint\n",
+        "\n",
+        "pp = pprint.PrettyPrinter(indent=4)\n",
+        "\n",
+        "\n",
         "SHARED_RESOURCE = \"projects/{project_id}/locations/{region}/deploymentResourcePools/{deployment_resource_pool_id}\".format(\n",
         "    project_id=PROJECT_ID,\n",
         "    region=REGION,\n",
@@ -1363,18 +1327,27 @@
         "    time.sleep(30)"
       ]
     },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "52248c450776"
+      },
+      "source": [
+        "### Get deployment details for the endpoint\n",
+        "\n",
+        "List the deployed models on the endpoint."
+      ]
+    },
     {
       "cell_type": "code",
       "execution_count": null,
       "metadata": {
-        "id": "86a659bf60f0"
+        "id": "3b768614e7c6"
       },
       "outputs": [],
       "source": [
-        "! curl -X GET \\\n",
-        " -H \"Authorization: Bearer $(gcloud auth print-access-token)\" \\\n",
-        " -H \"Content-Type: application/json\" \\\n",
-        "https://{REGION}-aiplatform.googleapis.com/v1/projects/759209241365/locations/us-central1/endpoints/2259566763823857664"
+        "print(endpoint_icn.list_models())\n",
+        "print(endpoint_use.list_models())"
       ]
     },
     {
@@ -1557,21 +1530,19 @@
       "source": [
         "#### Delete the `DeploymentResourcePool`\n",
         "\n",
-        "The method 'delete()' will delete your deployment resource pool."
+        "The method 'delete_deployment_resource_pool()' will delete your deployment resource pool."
       ]
     },
     {
       "cell_type": "code",
       "execution_count": null,
       "metadata": {
-        "id": "ac40cc1d594a"
+        "id": "b76a4de1e57e"
       },
       "outputs": [],
       "source": [
-        "! curl -X DELETE \\\n",
-        "-H \"Authorization: Bearer $(gcloud auth print-access-token)\" \\\n",
-        "-H \"Content-Type: application/json\" \\\n",
-        "https://{REGION}-aiplatform.googleapis.com/v1beta1/projects/{PROJECT_ID}/locations/{REGION}/deploymentResourcePools/{DEPLOYMENT_RESOURCE_POOL_ID}"
+        "response = pool_client.delete_deployment_resource_pool(name=deployment_pool_id)\n",
+        "print(response)"
       ]
     },
     {

notebooks/official/CODEOWNERS

@@ -42,3 +42,4 @@
 /sdk/SDK_FBProphet_Forecasting_Online.ipynb @brianchunkang
 /automl/sdk_automl_forecasting_hierarchical_batch.ipynb @ivanmkc
 /prediction/custom_batch_prediction_feature_filter.ipynb @soheilazangeneh
+/feature_store/feature_store_streaming_ingestion_sdk.ipynb @soheilazangeneh

notebooks/official/feature_store/feature_store_streaming_ingestion_sdk.ipynb

@@ -0,0 +1,774 @@
+{
+  "cells": [
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "ur8xi4C7S06n"
+      },
+      "outputs": [],
+      "source": [
+        "# Copyright 2022 Google LLC\n",
+        "#\n",
+        "# Licensed under the Apache License, Version 2.0 (the \"License\");\n",
+        "# you may not use this file except in compliance with the License.\n",
+        "# You may obtain a copy of the License at\n",
+        "#\n",
+        "#     https://www.apache.org/licenses/LICENSE-2.0\n",
+        "#\n",
+        "# Unless required by applicable law or agreed to in writing, software\n",
+        "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
+        "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
+        "# See the License for the specific language governing permissions and\n",
+        "# limitations under the License."
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "JAPoU8Sm5E6e"
+      },
+      "source": [
+        "# Feature Store: Streaming ingestion SDK\n",
+        "\n",
+        "<table align=\"left\">\n",
+        "\n",
+        "  <td>\n",
+        "    <a href=\"https://colab.research.google.com/github/GoogleCloudPlatform/vertex-ai-samples/blob/main/notebooks/official/feature_store/feature_store_streaming_ingestion_sdk.ipynb\">\n",
+        "      <img src=\"https://cloud.google.com/ml-engine/images/colab-logo-32px.png\" alt=\"Colab logo\"> Run in Colab\n",
+        "    </a>\n",
+        "  </td>\n",
+        "  <td>\n",
+        "    <a href=\"https://github.com/GoogleCloudPlatform/vertex-ai-samples/blob/main/notebooks/official/feature_store/feature_store_streaming_ingestion_sdk.ipynb\">\n",
+        "      <img src=\"https://cloud.google.com/ml-engine/images/github-logo-32px.png\" alt=\"GitHub logo\">\n",
+        "      View on GitHub\n",
+        "    </a>\n",
+        "  </td>\n",
+        "  <td>\n",
+        "    <a href=\"https://console.cloud.google.com/vertex-ai/workbench/deploy-notebook?download_url=https://raw.githubusercontent.com/GoogleCloudPlatform/vertex-ai-samples/main/notebooks/official/feature_store/feature_store_streaming_ingestion_sdk.ipynb\">\n",
+        "      <img src=\"https://lh3.googleusercontent.com/UiNooY4LUgW_oTvpsNhPpQzsstV5W8F7rYgxgGBD85cWJoLmrOzhVs_ksK_vgx40SHs7jCqkTkCk=e14-rj-sc0xffffff-h130-w32\" alt=\"Vertex AI logo\">\n",
+        "      Open in Vertex AI Workbench\n",
+        "    </a>\n",
+        "  </td>\n",
+        "</table>"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "24743cf4a1e1"
+      },
+      "source": [
+        "**_NOTE_**: This notebook has been tested in the following environment:\n",
+        "\n",
+        "* Python version = 3.9"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "tvgnzT1CKxrO"
+      },
+      "source": [
+        "## Overview\n",
+        "\n",
+        "This notebook demonstrates how to use Vertex AI Feature Store's streaming ingestion at the SDK layer."
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "d975e698c9a4"
+      },
+      "source": [
+        "### Objective\n",
+        "\n",
+        "In this tutorial, you learn how to ingest features from a `Pandas DataFrame` into your Vertex AI Feature Store using `write_feature_values` method from the Vertex AI SDK.\n",
+        "\n",
+        "This tutorial uses the following Google Cloud ML services and resources:\n",
+        "\n",
+        "- Vertex AI Feature Store\n",
+        "\n",
+        "\n",
+        "The steps performed include:\n",
+        "\n",
+        "- Create `Feature Store`\n",
+        "- Create new `Entity Type` for your `Feature Store`\n",
+        "- Ingest feature values from `Pandas DataFrame` into `Feature Store`'s `Entity Types`."
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "08d289fa873f"
+      },
+      "source": [
+        "### Dataset\n",
+        "\n",
+        "The dataset used for this notebook is the penguins dataset from [BigQuery public datasets](https://cloud.google.com/bigquery/public-data). This dataset has the following features: `culmen_length_mm`, `culmen_depth_mm`, `flipper_length_mm`, `body_mass_g`, `species`, and `sex`."
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "aed92deeb4a0"
+      },
+      "source": [
+        "### Costs\n",
+        "\n",
+        "This tutorial uses billable components of Google Cloud:\n",
+        "\n",
+        "* Vertex AI\n",
+        "\n",
+        "Learn about [Vertex AI\n",
+        "pricing](https://cloud.google.com/vertex-ai/pricing) and use the [Pricing\n",
+        "Calculator](https://cloud.google.com/products/calculator/)\n",
+        "to generate a cost estimate based on your projected usage.\n"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "i7EUnXsZhAGF"
+      },
+      "source": [
+        "## Installation\n",
+        "\n",
+        "Install the following packages required to execute this notebook."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "2b4ef9b72d43"
+      },
+      "outputs": [],
+      "source": [
+        "# Install the packages\n",
+        "! pip3 install --upgrade google-cloud-aiplatform\\\n",
+        "                         google-cloud-bigquery\\\n",
+        "                         numpy\\\n",
+        "                         pandas\\\n",
+        "                         pyarrow -q"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "58707a750154"
+      },
+      "source": [
+        "### Colab only: Uncomment the following cell to restart the kernel."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "f200f10a1da3"
+      },
+      "outputs": [],
+      "source": [
+        "# Automatically restart kernel after installs so that your environment can access the new packages\n",
+        "# import IPython\n",
+        "\n",
+        "# app = IPython.Application.instance()\n",
+        "# app.kernel.do_shutdown(True)"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "BF1j6f9HApxa"
+      },
+      "source": [
+        "## Before you begin\n",
+        "\n",
+        "### Set up your Google Cloud project\n",
+        "\n",
+        "**The following steps are required, regardless of your notebook environment.**\n",
+        "\n",
+        "1. [Select or create a Google Cloud project](https://console.cloud.google.com/cloud-resource-manager). When you first create an account, you get a $300 free credit towards your compute/storage costs.\n",
+        "\n",
+        "2. [Make sure that billing is enabled for your project](https://cloud.google.com/billing/docs/how-to/modify-project).\n",
+        "\n",
+        "3. [Enable the Vertex AI API](https://console.cloud.google.com/flows/enableapi?apiid=aiplatform.googleapis.com).\n",
+        "\n",
+        "4. If you are running this notebook locally, you need to install the [Cloud SDK](https://cloud.google.com/sdk)."
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "WReHDGG5g0XY"
+      },
+      "source": [
+        "#### Set your project ID\n",
+        "\n",
+        "**If you don't know your project ID**, try the following:\n",
+        "* Run `gcloud config list`.\n",
+        "* Run `gcloud projects list`.\n",
+        "* See the support page: [Locate the project ID](https://support.google.com/googleapi/answer/7014113)"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "oM1iC_MfAts1"
+      },
+      "outputs": [],
+      "source": [
+        "PROJECT_ID = \"[your-project-id]\"  # @param {type:\"string\"}\n",
+        "\n",
+        "# Set the project id\n",
+        "! gcloud config set project {PROJECT_ID}"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "region"
+      },
+      "source": [
+        "#### Region\n",
+        "\n",
+        "You can also change the `REGION` variable used by Vertex AI. Learn more about [Vertex AI regions](https://cloud.google.com/vertex-ai/docs/general/locations)."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "kljmKgilI_de"
+      },
+      "outputs": [],
+      "source": [
+        "REGION = \"us-central1\"  # @param {type: \"string\"}"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "sBCra4QMA2wR"
+      },
+      "source": [
+        "### Authenticate your Google Cloud account\n",
+        "\n",
+        "Depending on your Jupyter environment, you may have to manually authenticate. Follow the relevant instructions below."
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "74ccc9e52986"
+      },
+      "source": [
+        "**1. Vertex AI Workbench**\n",
+        "* Do nothing as you are already authenticated."
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "de775a3773ba"
+      },
+      "source": [
+        "**2. Local JupyterLab instance, uncomment and run:**"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "254614fa0c46"
+      },
+      "outputs": [],
+      "source": [
+        "# ! gcloud auth login"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "ef21552ccea8"
+      },
+      "source": [
+        "**3. Colab, uncomment and run:**"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "603adbbf0532"
+      },
+      "outputs": [],
+      "source": [
+        "# from google.colab import auth\n",
+        "# auth.authenticate_user()"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "f6b2ccc891ed"
+      },
+      "source": [
+        "**4. Service account or other**\n",
+        "* See how to grant Cloud Storage permissions to your service account at https://cloud.google.com/storage/docs/gsutil/commands/iam#ch-examples."
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "EsCYkJ4IU-z4"
+      },
+      "source": [
+        "### UUID\n",
+        "\n",
+        "If you are in a live tutorial session, you might be using a shared test account or project. To avoid name collisions between users on resources created, you create a uuid for each instance session, and append it onto the name of resources you create in this tutorial."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "4jWj2DSTU9my"
+      },
+      "outputs": [],
+      "source": [
+        "import random\n",
+        "import string\n",
+        "\n",
+        "\n",
+        "# Generate a uuid of a specifed length(default=8)\n",
+        "def generate_uuid(length: int = 8) -> str:\n",
+        "    return \"\".join(random.choices(string.ascii_lowercase + string.digits, k=length))\n",
+        "\n",
+        "\n",
+        "UUID = generate_uuid()"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "960505627ddf"
+      },
+      "source": [
+        "### Import libraries"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "PyQmSRbKA8r-"
+      },
+      "outputs": [],
+      "source": [
+        "import numpy as np\n",
+        "import pandas as pd\n",
+        "from google.cloud import aiplatform, bigquery"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "init_aip:mbsdk,all"
+      },
+      "source": [
+        "### Initialize Vertex AI SDK for Python\n",
+        "\n",
+        "Initialize the Vertex AI SDK for Python for your project."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "0ep8KuQhI_df"
+      },
+      "outputs": [],
+      "source": [
+        "aiplatform.init(project=PROJECT_ID, location=REGION)"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "k5XsEiAuEWUJ"
+      },
+      "source": [
+        "## Download and prepare the data"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "rOd7Ixa1pqBY"
+      },
+      "outputs": [],
+      "source": [
+        "def download_bq_table(bq_table_uri: str) -> pd.DataFrame:\n",
+        "    # Remove bq:// prefix if present\n",
+        "    prefix = \"bq://\"\n",
+        "    if bq_table_uri.startswith(prefix):\n",
+        "        bq_table_uri = bq_table_uri[len(prefix) :]\n",
+        "\n",
+        "    table = bigquery.TableReference.from_string(bq_table_uri)\n",
+        "\n",
+        "    # Create a BigQuery client\n",
+        "    bqclient = bigquery.Client(project=PROJECT_ID)\n",
+        "\n",
+        "    # Download the table rows\n",
+        "    rows = bqclient.list_rows(\n",
+        "        table,\n",
+        "    )\n",
+        "    return rows.to_dataframe()"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "SdX_m1Uppkfu"
+      },
+      "outputs": [],
+      "source": [
+        "BQ_SOURCE = \"bq://bigquery-public-data.ml_datasets.penguins\"\n",
+        "\n",
+        "# Download penguins BigQuery table\n",
+        "penguins_df = download_bq_table(BQ_SOURCE)"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "QuQe6mSbFbhm"
+      },
+      "source": [
+        "### Prepare the data\n",
+        "\n",
+        "Feature values to be written to the Feature Store can take the form of a list of `WriteFeatureValuesPayload` objects, a Python `dict` of the form\n",
+        "\n",
+        "`{entity_id : {feature_id : feature_value}, ...},`\n",
+        "\n",
+        "or a pandas `Dataframe`, where the `index` column holds the unique entity ID strings and each remaining column represents a feature.  In this notebook, since you use a pandas `DataFrame` for ingesting features we convert the index column data type to `string` to be used as `Entity ID`."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "cljxzJ3bqDer"
+      },
+      "outputs": [],
+      "source": [
+        "# Prepare the data\n",
+        "penguins_df.index = penguins_df.index.map(str)"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "GSxrSdSY2ovn"
+      },
+      "outputs": [],
+      "source": [
+        "# Remove null values\n",
+        "NA_VALUES = [\"NA\", \".\"]\n",
+        "penguins_df = penguins_df.replace(to_replace=NA_VALUES, value=np.NaN).dropna()"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "vgn4oQmSqdKI"
+      },
+      "source": [
+        "## Create Feature Store and define schemas\n",
+        "\n",
+        "Vertex AI Feature Store organizes resources hierarchically in the following order:\n",
+        "\n",
+        "`Featurestore -> EntityType -> Feature`\n",
+        "\n",
+        "You must create these resources before you can ingest data into Vertex AI Feature Store.\n",
+        "\n",
+        "Learn more about [Vertex AI Feature Store](https://cloud.google.com/vertex-ai/docs/featurestore)"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "yaHwdbGjZWTq"
+      },
+      "source": [
+        "### Create a Feature Store\n",
+        "\n",
+        "You create a Feature Store using `aiplatform.Featurestore.create` with the following parameters:\n",
+        "\n",
+        "* `featurestore_id (str)`: The ID to use for this Featurestore, which will become the final component of the Featurestore's resource name. The value must be unique within the project and location.\n",
+        "* `online_store_fixed_node_count`: Configuration for online serving resources.\n",
+        "* `project`: Project to create EntityType in. If not set, project set in `aiplatform.init` is used.\n",
+        "* `location`: Location to create EntityType in. If not set, location set in `aiplatform.init` is used.\n",
+        "* `sync`:  Whether to execute this creation synchronously."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "cImsONglqfxO"
+      },
+      "outputs": [],
+      "source": [
+        "FEATURESTORE_ID = f\"penguins_{UUID}\"\n",
+        "\n",
+        "penguins_feature_store = aiplatform.Featurestore.create(\n",
+        "    featurestore_id=FEATURESTORE_ID,\n",
+        "    online_store_fixed_node_count=1,\n",
+        "    project=PROJECT_ID,\n",
+        "    location=REGION,\n",
+        "    sync=True,\n",
+        ")"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "UfXgSD1VdzKb"
+      },
+      "source": [
+        "##### Verify that the Feature Store is created\n",
+        "Check if the Feature Store was successfully created by running the following code block."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "oud1OdfQd52r"
+      },
+      "outputs": [],
+      "source": [
+        "fs = aiplatform.Featurestore(\n",
+        "    featurestore_name=FEATURESTORE_ID,\n",
+        "    project=PROJECT_ID,\n",
+        "    location=REGION,\n",
+        ")\n",
+        "print(fs.gca_resource)"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "ep74rSlJWF3c"
+      },
+      "source": [
+        "### Create an EntityType\n",
+        "\n",
+        "An entity type is a collection of semantically related features. You define your own entity types, based on the concepts that are relevant to your use case. For example, a movie service might have the entity types `movie` and `user`, which group related features that correspond to movies or users.\n",
+        "\n",
+        "Here, you create an entity type entity type named `penguin_entity_type` using `create_entity_type` with the following parameters:\n",
+        "* `entity_type_id (str)`: The ID to use for the EntityType, which will become the final component of the EntityType's resource name. The value must be unique within a Feature Store.\n",
+        "* `description`: Description of the EntityType."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "zNzr-FlEr3tI"
+      },
+      "outputs": [],
+      "source": [
+        "ENTITY_TYPE_ID = f\"penguin_entity_type_{UUID}\"\n",
+        "\n",
+        "# Create penguin entity type\n",
+        "penguins_entity_type = penguins_feature_store.create_entity_type(\n",
+        "    entity_type_id=ENTITY_TYPE_ID,\n",
+        "    description=\"Penguins entity type\",\n",
+        ")"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "CquSdTp7duVw"
+      },
+      "source": [
+        "##### Verify that the EntityType is created\n",
+        "Check if the Entity Type was successfully created by running the following code block."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "76ocr_hJsG-t"
+      },
+      "outputs": [],
+      "source": [
+        "entity_type = penguins_feature_store.get_entity_type(entity_type_id=ENTITY_TYPE_ID)\n",
+        "\n",
+        "print(entity_type.gca_resource)"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "2vYV2UUFehwZ"
+      },
+      "source": [
+        "### Create Features\n",
+        "A feature is a measurable property or attribute of an entity type. For example, `penguin` entity type has features such as `flipper_length_mm`, and `body_mass_g`. Features can be created within each entity type.\n",
+        "\n",
+        "When you create a feature, you specify its value type such as `DOUBLE`, and `STRING`. This value determines what value types you can ingest for a particular feature.\n",
+        "\n",
+        "Learn more about [Feature Value Types](https://cloud.google.com/vertex-ai/docs/reference/rest/v1/projects.locations.featurestores.entityTypes.features)"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "WQ5EsPPbsSuE"
+      },
+      "outputs": [],
+      "source": [
+        "penguins_feature_configs = {\n",
+        "    \"species\": {\n",
+        "        \"value_type\": \"STRING\",\n",
+        "    },\n",
+        "    \"island\": {\n",
+        "        \"value_type\": \"STRING\",\n",
+        "    },\n",
+        "    \"culmen_length_mm\": {\n",
+        "        \"value_type\": \"DOUBLE\",\n",
+        "    },\n",
+        "    \"culmen_depth_mm\": {\n",
+        "        \"value_type\": \"DOUBLE\",\n",
+        "    },\n",
+        "    \"flipper_length_mm\": {\n",
+        "        \"value_type\": \"DOUBLE\",\n",
+        "    },\n",
+        "    \"body_mass_g\": {\"value_type\": \"DOUBLE\"},\n",
+        "    \"sex\": {\"value_type\": \"STRING\"},\n",
+        "}"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "AKRXJCPijM8w"
+      },
+      "source": [
+        "You can create features either using `create_feature` or `batch_create_features`. Here, for convinience, you have added all feature configs in one variabel, so we use `batch_create_features`."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "tXOI1Onhs46x"
+      },
+      "outputs": [],
+      "source": [
+        "penguin_features = penguins_entity_type.batch_create_features(\n",
+        "    feature_configs=penguins_feature_configs,\n",
+        ")"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "WBx26pZItUN4"
+      },
+      "source": [
+        "### Write features to the Feature Store\n",
+        "Use the `write_feature_values` API to write a feature to the Feature Store with the following parameter:\n",
+        "\n",
+        "* `instances`: Feature values to be written to the Feature Store that can take the form of a list of WriteFeatureValuesPayload objects, a Python dict, or a pandas Dataframe.\n",
+        "\n",
+        "This streaming ingestion feature has been introduced to the Vertex AI SDK under the **preview** namespace. Here, you pass the pandas `Dataframe` you created from penguins dataset as `instances` parameter.\n",
+        "\n",
+        "Learn more about [Streaming ingestion API](https://github.com/googleapis/python-aiplatform/blob/e6933503d2d3a0f8a8f7ef8c178ed50a69ac2268/google/cloud/aiplatform/preview/featurestore/entity_type.py#L36)"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "iUGI-ftltXqE"
+      },
+      "outputs": [],
+      "source": [
+        "penguins_entity_type.preview.write_feature_values(instances=penguins_df)"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "STq67KHO3q_e"
+      },
+      "source": [
+        "## Read back written features\n",
+        "\n",
+        "Wait a few seconds for the write to propagate, then do an online read to confirm the write was successful."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "lwoMnze43r9G"
+      },
+      "outputs": [],
+      "source": [
+        "ENTITY_IDS = [str(x) for x in range(100)]\n",
+        "penguins_entity_type.read(entity_ids=ENTITY_IDS)"
+      ]
+    },
+    {
+      "cell_type": "markdown",
+      "metadata": {
+        "id": "TpV-iwP9qw9c"
+      },
+      "source": [
+        "## Cleaning up\n",
+        "\n",
+        "To clean up all Google Cloud resources used in this project, you can [delete the Google Cloud\n",
+        "project](https://cloud.google.com/resource-manager/docs/creating-managing-projects#shutting_down_projects) you used for the tutorial.\n",
+        "\n",
+        "Otherwise, you can delete the individual resources you created in this tutorial:"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {
+        "id": "sx_vKniMq9ZX"
+      },
+      "outputs": [],
+      "source": [
+        "penguins_feature_store.delete(force=True)"
+      ]
+    }
+  ],
+  "metadata": {
+    "colab": {
+      "name": "feature_store_streaming_ingestion_sdk.ipynb",
+      "toc_visible": true
+    },
+    "kernelspec": {
+      "display_name": "Python 3",
+      "name": "python3"
+    }
+  },
+  "nbformat": 4,
+  "nbformat_minor": 0
+}

notebooks/official/tabular_workflows/tabnet_on_vertex_pipelines.ipynb

@@ -97,7 +97,7 @@
         "### Dataset\n",
         "\n",
         "The dataset you will be using is [Bank Marketing](https://archive.ics.uci.edu/ml/datasets/bank+marketing).\n",
-        "The data is for direct marketing campaigns (phone calls) of a Portuguese banking institution. The binary classification goal is to predict if a client subscribe a term deposit. For this notebook, you randomly selected 90% of the rows in the original dataset and saved them in a train.csv file hosted on Cloud Storage. To download the file, click [here](https://storage.googleapis.com/cloud-samples-data/vertex-ai/tabular-workflows/datasets/bank-marketing/train.csv)."
+        "The data is for direct marketing campaigns (phone calls) of a Portuguese banking institution. The binary classification goal is to predict if a client subscribe a term deposit. For this notebook, you randomly selected 90% of the rows in the original dataset and saved them in a train.csv file hosted on Cloud Storage. To download the file, click [here](https://storage.googleapis.com/cloud-samples-data-us-central1/vertex-ai/tabular-workflows/datasets/bank-marketing/train.csv)."
       ]
     },
     {
@@ -746,7 +746,7 @@
       },
       "outputs": [],
       "source": [
-        "data_source_csv_filenames = \"gs://cloud-samples-data/vertex-ai/tabular-workflows/datasets/bank-marketing/train.csv\"\n",
+        "data_source_csv_filenames = \"gs://cloud-samples-data-us-central1/vertex-ai/tabular-workflows/datasets/bank-marketing/train.csv\"\n",
         "data_source_bigquery_table_path = (\n",
         "    None  # @param {type:\"string\"}, format: bq://bq_project.bq_dataset.bq_table\n",
         ")"

notebooks/official/tabular_workflows/wide_and_deep_on_vertex_pipelines.ipynb

@@ -97,7 +97,7 @@
         "### Dataset\n",
         "\n",
         "The dataset you will be using is [Bank Marketing](https://archive.ics.uci.edu/ml/datasets/bank+marketing).\n",
-        "The data is for direct marketing campaigns (phone calls) of a Portuguese banking institution. The binary classification goal is to predict if a client will subscribe a term deposit. For this notebook, we randomly selected 90% of the rows in the original dataset and saved them in a train.csv file hosted on Cloud Storage. To download the file, click [here](https://storage.googleapis.com/cloud-samples-data/vertex-ai/tabular-workflows/datasets/bank-marketing/train.csv)."
+        "The data is for direct marketing campaigns (phone calls) of a Portuguese banking institution. The binary classification goal is to predict if a client will subscribe a term deposit. For this notebook, we randomly selected 90% of the rows in the original dataset and saved them in a train.csv file hosted on Cloud Storage. To download the file, click [here](https://storage.googleapis.com/cloud-samples-data-us-central1/vertex-ai/tabular-workflows/datasets/bank-marketing/train.csv)."
       ]
     },
     {
@@ -667,7 +667,7 @@
       },
       "outputs": [],
       "source": [
-        "data_source_csv_filenames = \"gs://cloud-samples-data/vertex-ai/tabular-workflows/datasets/bank-marketing/train.csv\"\n",
+        "data_source_csv_filenames = \"gs://cloud-samples-data-us-central1/vertex-ai/tabular-workflows/datasets/bank-marketing/train.csv\"\n",
         "data_source_bigquery_table_path = (\n",
         "    None  # @param {type:\"string\"}, format: bq://bq_project.bq_dataset.bq_table\n",
         ")"

notebooks/official/test.ipynb

@@ -0,0 +1,61 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "346219a0-c421-4986-89c0-1cab5f9896bc",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import time"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "fdeaadd2-900d-4580-a637-b0ec3c892ce7",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "time.sleep(5 * 60)\n",
+    "print(\"5 mins\")\n",
+    "\n",
+    "time.sleep(5 * 60)\n",
+    "print(\"10 mins\")\n",
+    "\n",
+    "time.sleep(5 * 60)\n",
+    "print(\"15 mins\")\n",
+    "\n",
+    "time.sleep(5 * 60)\n",
+    "print(\"20 mins\")"
+   ]
+  }
+ ],
+ "metadata": {
+  "environment": {
+   "kernel": "python3",
+   "name": "common-cpu.m95",
+   "type": "gcloud",
+   "uri": "gcr.io/deeplearning-platform-release/base-cpu:m95"
+  },
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.7.12"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}