Differential D8684 Diff 40807 examples/docs_snippets/docs_snippets/concepts/assets/materialization_solids.py

Changeset View

Standalone View

examples/docs_snippets/docs_snippets/concepts/assets/materialization_solids.py

	Show All 37 Lines

	# end_materialization_solids_marker_0			# end_materialization_solids_marker_0

	# start_materialization_solids_marker_1			# start_materialization_solids_marker_1
	@solid			@solid
	def my_materialization_solid(context):			def my_materialization_solid(context):
	df = read_df()			df = read_df()
	remote_storage_path = persist_to_storage(df)			remote_storage_path = persist_to_storage(df)
	yield AssetMaterialization(asset_key="my_dataset", description="Persisted result to storage")			yield AssetMaterialization(
				asset_key="my_dataset", description="Persisted result to storage"
				)
	yield Output(remote_storage_path)			yield Output(remote_storage_path)


	# end_materialization_solids_marker_1			# end_materialization_solids_marker_1

	# start_simple_asset_solid			# start_simple_asset_solid


	Show All 39 Lines
	# start_partitioned_asset_materialization			# start_partitioned_asset_materialization


	@solid(config_schema={"date": str})			@solid(config_schema={"date": str})
	def my_partitioned_asset_solid(context):			def my_partitioned_asset_solid(context):
	partition_date = context.solid_config["date"]			partition_date = context.solid_config["date"]
	df = read_df_for_date(partition_date)			df = read_df_for_date(partition_date)
	remote_storage_path = persist_to_storage(df)			remote_storage_path = persist_to_storage(df)
	yield AssetMaterialization(asset_key="my_dataset", partition=partition_date)			yield AssetMaterialization(
				asset_key="my_dataset", partition=partition_date
				)
	yield Output(remote_storage_path)			yield Output(remote_storage_path)


	# end_partitioned_asset_materialization			# end_partitioned_asset_materialization


	# start_materialization_solids_marker_2			# start_materialization_solids_marker_2
	@solid			@solid
	def my_metadata_materialization_solid(context):			def my_metadata_materialization_solid(context):
	df = read_df()			df = read_df()
	remote_storage_path = persist_to_storage(df)			remote_storage_path = persist_to_storage(df)
	yield AssetMaterialization(			yield AssetMaterialization(
	asset_key="my_dataset",			asset_key="my_dataset",
	description="Persisted result to storage",			description="Persisted result to storage",
	metadata={			metadata={
	"text_metadata": "Text-based metadata for this event",			"text_metadata": "Text-based metadata for this event",
	"path": EventMetadata.path(remote_storage_path),			"path": EventMetadata.path(remote_storage_path),
	"dashboard_url": EventMetadata.url("http://mycoolsite.com/url_for_my_data"),			"dashboard_url": EventMetadata.url(
				"http://mycoolsite.com/url_for_my_data"
				),
	"size (bytes)": calculate_bytes(df),			"size (bytes)": calculate_bytes(df),
	},			},
	)			)
	yield Output(remote_storage_path)			yield Output(remote_storage_path)


	# end_materialization_solids_marker_2			# end_materialization_solids_marker_2


	# start_materialization_solids_marker_3			# start_materialization_solids_marker_3
	@solid			@solid
	def my_asset_key_materialization_solid(context):			def my_asset_key_materialization_solid(context):
	df = read_df()			df = read_df()
	remote_storage_path = persist_to_storage(df)			remote_storage_path = persist_to_storage(df)
	yield AssetMaterialization(			yield AssetMaterialization(
	asset_key=AssetKey(["dashboard", "my_cool_site"]),			asset_key=AssetKey(["dashboard", "my_cool_site"]),
	description="Persisted result to storage",			description="Persisted result to storage",
	metadata={			metadata={
	"dashboard_url": EventMetadata.url("http://mycoolsite.com/dashboard"),			"dashboard_url": EventMetadata.url(
				"http://mycoolsite.com/dashboard"
				),
	"size (bytes)": calculate_bytes(df),			"size (bytes)": calculate_bytes(df),
	},			},
	)			)
	yield Output(remote_storage_path)			yield Output(remote_storage_path)


	# end_materialization_solids_marker_3			# end_materialization_solids_marker_3


	@pipeline			@pipeline
	def my_asset_pipeline():			def my_asset_pipeline():
	my_materialization_solid()			my_materialization_solid()