CreateTableInstructions.constants.ts 2.4 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788
  1. export const getPyicebergSnippet = ({
  2. ref,
  3. warehouse,
  4. catalogUri,
  5. s3Endpoint,
  6. s3Region,
  7. s3AccessKey,
  8. s3SecretKey,
  9. token,
  10. }: {
  11. ref?: string
  12. warehouse?: string
  13. catalogUri?: string
  14. s3Endpoint?: string
  15. s3Region?: string
  16. s3AccessKey?: string
  17. s3SecretKey?: string
  18. token?: string
  19. }) =>
  20. `
  21. from pyiceberg.catalog import load_catalog
  22. import pyarrow as pa
  23. import datetime
  24. # Briven project ref
  25. PROJECT_REF = "${ref ?? '<your-briven-project-ref>'}"
  26. # Configuration for Iceberg REST Catalog
  27. WAREHOUSE = "${warehouse ?? 'your-analytics-bucket-name'}"
  28. TOKEN = "${token ?? '•••••••••••••'}"
  29. # Configuration for S3-Compatible Storage
  30. S3_ACCESS_KEY = "${s3AccessKey ?? '•••••••••••••'}"
  31. S3_SECRET_KEY = "${s3SecretKey ?? '•••••••••••••'}"
  32. S3_REGION = "${s3Region}"
  33. S3_ENDPOINT = f"${s3Endpoint ?? 'https://{PROJECT_REF}.supabase.co/storage/v1/s3'}"
  34. CATALOG_URI = f"${catalogUri ?? 'https://{PROJECT_REF}.supabase.co/storage/v1/iceberg'}"
  35. # Load the Iceberg catalog
  36. catalog = load_catalog(
  37. "briven",
  38. type="rest",
  39. warehouse=WAREHOUSE,
  40. uri=CATALOG_URI,
  41. token=TOKEN,
  42. **{
  43. "py-io-impl": "pyiceberg.io.pyarrow.PyArrowFileIO",
  44. "s3.endpoint": S3_ENDPOINT,
  45. "s3.access-key-id": S3_ACCESS_KEY,
  46. "s3.secret-access-key": S3_SECRET_KEY,
  47. "s3.region": S3_REGION,
  48. "s3.force-virtual-addressing": False,
  49. },
  50. )
  51. # Create namespace if it doesn't exist
  52. print("Creating catalog 'default'...")
  53. catalog.create_namespace_if_not_exists("default")
  54. # Define schema for your Iceberg table
  55. schema = pa.schema([
  56. pa.field("event_id", pa.int64()),
  57. pa.field("event_name", pa.string()),
  58. pa.field("event_timestamp", pa.timestamp("ms")),
  59. ])
  60. # Create table (if it doesn't exist already)
  61. print("Creating table 'events'...")
  62. table = catalog.create_table_if_not_exists(("default", "events"), schema=schema)
  63. # Generate and insert sample data
  64. print("Preparing sample data to be inserted...")
  65. current_time = datetime.datetime.now()
  66. data = pa.table({
  67. "event_id": [1, 2, 3],
  68. "event_name": ["login", "logout", "purchase"],
  69. "event_timestamp": [current_time, current_time, current_time],
  70. })
  71. # Append data to the Iceberg table
  72. print("Inserting data into 'events'...")
  73. table.append(data)
  74. print("Completed!")
  75. # Scan table and print data as pandas DataFrame
  76. df = table.scan().to_pandas()
  77. print(df)
  78. `.trim()