code: def test_feature_processor_transform_offline_only_store_ingestion_run_with_remote(
      sagemaker_session,
  ):
      car_data_feature_group_name = get_car_data_feature_group_name()
      car_data_aggregated_feature_group_name = (
          get_car_data_aggregated_feature_group_name()
      )
      try:
          feature_groups = create_feature_groups(
              sagemaker_session=sagemaker_session,
              car_data_feature_group_name=car_data_feature_group_name,
              car_data_aggregated_feature_group_name=car_data_aggregated_feature_group_name,
              offline_store_s3_uri=get_offline_store_s3_uri(
                  sagemaker_session=sagemaker_session
              ),
          )
  
          raw_data_uri = get_raw_car_data_s3_uri(
              sagemaker_session=sagemaker_session
          )
          whl_file_uri = get_wheel_file_s3_uri(
              sagemaker_session=sagemaker_session
          )
          whl_file_name = os.path.basename(whl_file_uri)
  
          pre_execution_commands = [
              f"aws s3 cp {whl_file_uri} ./",
              f"/usr/local/bin/python3.9 -m pip install ./{whl_file_name} --force-reinstall",
          ]
  
          @remote(
              pre_execution_commands=pre_execution_commands,
              spark_config=SparkConfig(),
              instance_type="ml.m5.xlarge",
          )
          @feature_processor(
              inputs=[CSVDataSource(raw_data_uri)],
              output=feature_groups["car_data_arn"],
              target_stores=["OfflineStore"],
          )
          def transform(raw_s3_data_as_df):
              """Load data from S3, perform basic feature engineering, store it in a Feature Group"""
              from pyspark.sql.functions import regexp_replace
              from pyspark.sql.functions import lit
  
              transformed_df = (
                  raw_s3_data_as_df
                  # Rename Columns
                  .withColumnRenamed("Id", "id")
                  .withColumnRenamed("Model", "model")
                  .withColumnRenamed("Year", "model_year")
                  .withColumnRenamed("Status", "status")
                  .withColumnRenamed("Mileage", "mileage")
                  .withColumnRenamed("Price", "price")
                  .withColumnRenamed("MSRP", "msrp")
                  # Add Event Time
                  .withColumn("ingest_time", lit(int(time.time())))
                  # Remove punctuation and fluff; replace with NA
                  .withColumn(
                      "Price", regexp_replace("Price", "\$", "")
                  )  # noqa: W605
                  .withColumn(
                      "mileage",
                      regexp_replace("mileage", "(,)|(mi\.)", ""),
                  )  # noqa: W605
                  .withColumn(
                      "mileage",
                      regexp_replace("mileage", "Not available", "NA"),
                  )
                  .withColumn("price", regexp_replace("price", ",", ""))
                  .withColumn(
                      "msrp",
                      regexp_replace("msrp", "(^MSRP\s\\$)|(,)", ""),
                  )  # noqa: W605
                  .withColumn(
                      "msrp",
                      regexp_replace("msrp", "Not specified", "NA"),
                  )
                  .withColumn(
                      "msrp",
                      regexp_replace("msrp", "\\$\d+[a-zA-Z\s]+", "NA"),
                  )  # noqa: W605
                  .withColumn(
                      "model",
                      regexp_replace("model", "^\d\d\d\d\s", ""),
                  )  # noqa: W605
              )
  
              transformed_df.show()
              return transformed_df
  
          transform()
  
          featurestore_client = (
              sagemaker_session.sagemaker_featurestore_runtime_client
          )
          results = featurestore_client.batch_get_record(
              Identifiers=[
                  {
                      "FeatureGroupName": car_data_feature_group_name,
                      "RecordIdentifiersValueAsString": [
                          "0",
                          "1",
                          "2",
                          "3",
                          "4",
                          "5",
                          "6",
                          "7",
                          "8",
                          "9",
                          "10",
                          "11",
                          "12",
                          "13",
                          "14",
                          "15",
                          "16",
                          "17",
                          "18",
                          "19",
                          "20",
                          "21",
                          "22",
                          "23",
                          "24",
                          "25",
                      ],
                  },
              ]
          )
  
          assert len(results["Records"]) == 0
  
          car_sales_query = feature_groups[
              "car_data_feature_group"
          ].athena_query()
          query = f'SELECT * FROM "sagemaker_featurestore".{car_sales_query.table_name} LIMIT 1000;'
          output_uri = "s3://{}/{}/input/data/{}".format(
              sagemaker_session.default_bucket(),
              "feature-processor-test",
              "csv-data-fg-result",
          )
          car_sales_query.run(
              query_string=query, output_location=output_uri
          )
          car_sales_query.wait()
          dataset = car_sales_query.as_dataframe()
          dataset = dataset.drop(
              columns=[
                  "ingest_time",
                  "write_time",
                  "api_invocation_time",
                  "is_deleted",
              ]
          )
  
          assert dataset.equals(get_expected_dataframe())
      finally:
          cleanup_offline_store(
              feature_group=feature_groups["car_data_feature_group"],
              sagemaker_session=sagemaker_session,
          )
          cleanup_offline_store(
              feature_group=feature_groups[
                  "car_data_aggregated_feature_group"
              ],
              sagemaker_session=sagemaker_session,
          )
          cleanup_feature_group(
              feature_groups["car_data_feature_group"],
              sagemaker_session=sagemaker_session,
          )
          cleanup_feature_group(
              feature_groups["car_data_aggregated_feature_group"],
              sagemaker_session=sagemaker_session,
          )
  
imports: from __future__ import absolute_import
  import logging
  import os
  import subprocess
  import sys
  import time
  from typing import Dict
  from datetime import datetime
  from pyspark.sql import DataFrame
  import pytz
  import pytest
  import pandas
  import numpy
  import json
  import attr
  from boto3 import client
  from tests.integ import DATA_DIR
  from sagemaker.session import Session
  from sagemaker.s3 import S3Uploader
  from urllib.parse import urlparse
  from sagemaker import get_execution_role
  from sagemaker.remote_function import remote
  from sagemaker.remote_function.spark_config import SparkConfig
  from sagemaker.feature_store.feature_definition import FeatureDefinition, FeatureTypeEnum
  from sagemaker.feature_store.feature_group import FeatureGroup
  from sagemaker.feature_store.feature_processor import feature_processor, CSVDataSource, PySparkDataSource, FeatureProcessorPipelineEvents, FeatureProcessorPipelineExecutionStatus
  from sagemaker.feature_store.feature_processor.feature_scheduler import to_pipeline, describe, execute, schedule, delete_schedule, put_trigger, enable_trigger, disable_trigger, delete_trigger
  from sagemaker.workflow.pipeline import Pipeline
label_desc: Use image node_id 4 for calling a global function with example usage: test_feature_processor_transform_offline_only_store_ingestion_run_with_remote(sagemaker_session) and returns: transformed_df