code: def create_test_scalar_dataset(
      output_url, num_rows, num_files=4, spark=None, partition_by=None
  ):
      """Creates a dataset in tmp_url location. The dataset emulates non-petastorm dataset, i.e. contains only native
      parquet types.
  
      These are the fields with mock data:
        'id', 'int_fixed_size_list', 'datetime', 'timestamp', 'string', 'string2', 'float64'
  
      :param output_url: Url specifying the location the parquet store is written to
      :param num_rows: Number of rows in the generated dataset
      :param num_files: Number of parquet files that will be written into the parquet store
      :param spark: An instance of spark session object. If `None` (default), a new spark session is created.
      :param partition_by: A list of fields to partition the parquet store by.
      :return: A list of records with a copy of the data written to the dataset.
      """
  
      partition_by = partition_by or []
      shutdown = False
      if not spark:
          spark_session = SparkSession.builder.appName(
              "petastorm_end_to_end_test"
          ).master("local[*]")
  
          spark = spark_session.getOrCreate()
          shutdown = True
          hadoop_config = spark.sparkContext._jsc.hadoopConfiguration()
          hadoop_config.setInt("parquet.block.size", 100)
  
      def expected_row(i):
          result = {
              "id": np.int32(i),
              "id_div_700": np.int32(i // 700),
              "datetime": np.datetime64("2019-01-02"),
              "timestamp": np.datetime64("2005-02-25T03:30"),
              "string": np.unicode_("hello_{}".format(i)),
              "string2": np.unicode_("world_{}".format(i)),
              "float64": np.float64(i) * 0.66,
              "int_fixed_size_list": np.arange(1 + i, 10 + i).astype(
                  np.int32
              ),
          }
          return result
  
      expected_data = [expected_row(i) for i in range(num_rows)]
  
      expected_data_as_scalars = [
          {
              k: v.item() if isinstance(v, np.generic) else v
              for k, v in row.items()
          }
          for row in expected_data
      ]
  
      # np.datetime64 is converted to a timezone unaware datetime instances. Working explicitly in UTC so we don't need
      # to think about local timezone in the tests
      for row in expected_data_as_scalars:
          row["timestamp"] = row["timestamp"].replace(tzinfo=pytz.UTC)
          row["int_fixed_size_list"] = row[
              "int_fixed_size_list"
          ].tolist()
          row["nested_struct"] = {"nested_int": 0}
  
      rows = [
          Row(
              **OrderedDict(
                  sorted(row.items(), key=lambda item: item[0])
              )
          )
          for row in expected_data_as_scalars
      ]
  
      maybe_int_fixed_size_list_field = [
          StructField(
              "int_fixed_size_list",
              ArrayType(IntegerType(), False),
              False,
          )
      ]
  
      # WARNING: surprisingly, schema fields and row fields are matched only by order and not name.
      # We must maintain alphabetical order of the struct fields for the code to work!!!
      schema = StructType(
          [
              StructField("datetime", DateType(), False),
              StructField("float64", DoubleType(), False),
              StructField("id", IntegerType(), False),
              StructField("id_div_700", IntegerType(), False),
          ]
          + maybe_int_fixed_size_list_field
          + [
              StructField(
                  "nested_struct",
                  StructType(
                      [StructField("nested_int", IntegerType(), True)]
                  ),
              ),
              StructField("string", StringType(), False),
              StructField("string2", StringType(), False),
              StructField("timestamp", TimestampType(), False),
          ]
      )
  
      dataframe = spark.createDataFrame(rows, schema)
      dataframe.coalesce(num_files).write.option(
          "compression", "none"
      ).mode("overwrite").partitionBy(*partition_by).parquet(output_url)
  
      if shutdown:
          spark.stop()
  
      return expected_data
  
imports: from __future__ import division
  import random
  from collections import OrderedDict
  from decimal import Decimal
  from functools import partial
  import numpy
  import pytz
  from pyspark import Row
  from pyspark.sql import SparkSession
  from pyspark.sql.types import StringType, DecimalType, DoubleType, StructField, IntegerType, StructType, DateType, TimestampType, ShortType, ArrayType
  from petastorm.codecs import CompressedImageCodec, NdarrayCodec, ScalarCodec
  from petastorm.etl.dataset_metadata import materialize_dataset
  from petastorm.etl.rowgroup_indexers import SingleFieldIndexer
  from petastorm.etl.rowgroup_indexing import build_rowgroup_index
  from petastorm.unischema import Unischema, UnischemaField, dict_to_spark_row
label_desc: Use image node_id 5 for calling a global function with example usage: create_test_scalar_dataset(output_url, num_rows, num_files, spark, partition_by) and returns: expected_data, result